From b28bf785a64278464d367f3bf52f650b62feb241 Mon Sep 17 00:00:00 2001 From: Dustin Byrne Date: Fri, 30 Oct 2026 11:05:01 -0400 Subject: [PATCH 1/3] feat: add paired Ruby SDK performance benchmarks --- .github/workflows/benchmark-report.yml | 21 ++ .github/workflows/benchmarks.yml | 149 +++++++++ CONTRIBUTING.md | 31 ++ benchmarks/.gitignore | 2 + benchmarks/check_native.py | 97 ++++++ benchmarks/controls.rb | 42 +++ benchmarks/fixtures.py | 216 +++++++++++++ benchmarks/prepare.py | 85 +++++ benchmarks/run.py | 422 +++++++++++++++++++++++++ benchmarks/server.py | 159 ++++++++++ benchmarks/shards.json | 39 +++ benchmarks/test_benchmarks.py | 343 ++++++++++++++++++++ benchmarks/worker.rb | 291 +++++++++++++++++ 13 files changed, 1897 insertions(+) create mode 100644 .github/workflows/benchmark-report.yml create mode 100644 .github/workflows/benchmarks.yml create mode 100644 benchmarks/.gitignore create mode 100644 benchmarks/check_native.py create mode 100644 benchmarks/controls.rb create mode 100644 benchmarks/fixtures.py create mode 100644 benchmarks/prepare.py create mode 100644 benchmarks/run.py create mode 100644 benchmarks/server.py create mode 100644 benchmarks/shards.json create mode 100644 benchmarks/test_benchmarks.py create mode 100644 benchmarks/worker.rb diff --git a/.github/workflows/benchmark-report.yml b/.github/workflows/benchmark-report.yml new file mode 100644 index 00000000..6de61dc9 --- /dev/null +++ b/.github/workflows/benchmark-report.yml @@ -0,0 +1,21 @@ +name: Ruby SDK benchmark report + +on: + workflow_run: + workflows: ['Ruby SDK benchmarks'] + types: [completed] + +permissions: {} + +jobs: + report: + permissions: + contents: read + actions: read + pull-requests: write + uses: PostHog/.github/.github/workflows/sdk-benchmark-report.yml@2a6111d08d784c0ca84f7f97f32a13fdd6d31449 # SDK benchmark workflow (PostHog/.github PR #84) + with: + workflow-path: .github/workflows/benchmarks.yml + artifact-name: ruby-benchmark-results + report-key: posthog-ruby + sdk: posthog-ruby diff --git a/.github/workflows/benchmarks.yml b/.github/workflows/benchmarks.yml new file mode 100644 index 00000000..061dedfc --- /dev/null +++ b/.github/workflows/benchmarks.yml @@ -0,0 +1,149 @@ +name: Ruby SDK benchmarks + +on: [pull_request] + +permissions: {} + +concurrency: + group: ${{ github.workflow }}-${{ github.event.pull_request.number }} + cancel-in-progress: true + +jobs: + benchmark: + runs-on: ubuntu-24.04 + timeout-minutes: 12 + env: + BUNDLE_PATH: ${{ github.workspace }}/vendor/bundle + strategy: + fail-fast: false + matrix: + shard: [flags-single, flags-scaling, flags-memory, capture] + permissions: + contents: read + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.event.pull_request.head.sha }} + fetch-depth: 0 + persist-credentials: false + + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: '3.12.12' + + - uses: ruby/setup-ruby@b1da4809b005e0178fe08a1d57317bffc00b7b4c # v1.332.0 + with: + ruby-version: '3.4.7' + bundler: '4.0.13' + bundler-cache: true + + - name: Prepare pinned baseline + env: + TARGET_SHA: ${{ github.event.pull_request.base.sha }} + HEAD_SHA: ${{ github.event.pull_request.head.sha }} + run: | + set -euo pipefail + export BASE_WORKTREE="$RUNNER_TEMP/ruby-benchmark-base" + export RESULTS="$RUNNER_TEMP/ruby-benchmark-results" + printf 'BASE_WORKTREE=%s\nRESULTS=%s\n' "$BASE_WORKTREE" "$RESULTS" >> "$GITHUB_ENV" + BASE_SHA=$(git merge-base "$TARGET_SHA" "$HEAD_SHA") + git worktree add --detach "$BASE_WORKTREE" "$BASE_SHA" + mkdir -p "$RESULTS" + python -c 'import json, os, pathlib, sys; pathlib.Path(os.environ["RESULTS"], "revisions.json").write_text(json.dumps({"target":os.environ["TARGET_SHA"], "head":os.environ["HEAD_SHA"], "base":sys.argv[1]}, indent=2))' "$BASE_SHA" + + - name: Build and install isolated gems + run: | + python -B benchmarks/prepare.py . "$RUNNER_TEMP/ruby-benchmark-build/head" + python -B benchmarks/prepare.py "$BASE_WORKTREE" "$RUNNER_TEMP/ruby-benchmark-build/base" + + - name: Test measurement controls and fixtures + run: | + python -B benchmarks/test_benchmarks.py + python -B benchmarks/check_native.py "$RUNNER_TEMP/ruby-benchmark-build/head" + + - name: Measure paired base and head + id: measure + timeout-minutes: 10 + run: | + python -B benchmarks/run.py \ + --base "$BASE_WORKTREE" \ + --head "$GITHUB_WORKSPACE" \ + --build-directory "$RUNNER_TEMP/ruby-benchmark-build" \ + --output "$RESULTS" \ + --repetitions 6 \ + --shard "${{ matrix.shard }}" + + - name: Upload raw shard results + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: benchmark-shard-${{ matrix.shard }} + path: | + ${{ runner.temp }}/ruby-benchmark-results/base.json + ${{ runner.temp }}/ruby-benchmark-results/head.json + ${{ runner.temp }}/ruby-benchmark-results/revisions.json + if-no-files-found: warn + retention-days: 14 + + collect: + needs: benchmark + if: ${{ !cancelled() }} + runs-on: ubuntu-24.04 + timeout-minutes: 5 + permissions: + contents: read + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.event.pull_request.head.sha }} + persist-credentials: false + + - name: Download shard results + uses: actions/download-artifact@9000827ccba6bdab643e8b6fd33ac0654aef8333 # v8.0.2 + with: + pattern: benchmark-shard-* + path: ${{ runner.temp }}/ruby-benchmark-shards + merge-multiple: false + + - name: Collect and render benchmark comparison + uses: PostHog/.github/.github/actions/benchmark-report@04f8421ec4daa7f4066eaeb4c53621ba79ddc3ca # SDK benchmark reporting (PostHog/.github PR #84) + with: + mode: collect + source-status: ${{ needs.benchmark.result }} + shard-manifest: benchmarks/shards.json + shard-directory: ${{ runner.temp }}/ruby-benchmark-shards + output-directory: ${{ runner.temp }}/ruby-benchmark-report + + - name: Upload rendered comparison + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: ruby-benchmark-report + path: | + ${{ runner.temp }}/ruby-benchmark-report/report.md + ${{ runner.temp }}/ruby-benchmark-report/comparison.json + if-no-files-found: ignore + retention-days: 14 + + - name: Upload raw benchmark results + if: always() + id: results + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: ruby-benchmark-results + path: | + ${{ runner.temp }}/ruby-benchmark-report/bundle.json + ${{ runner.temp }}/ruby-benchmark-report/revisions.json + if-no-files-found: warn + retention-days: 14 + + - name: Link results + if: always() + env: + ARTIFACT_URL: ${{ steps.results.outputs.artifact-url }} + run: | + printf '## Ruby SDK benchmark data\n\n' >> "$GITHUB_STEP_SUMMARY" + if [[ -n "$ARTIFACT_URL" ]]; then + printf '[Download paired measurements and execution metadata](%s)\n\n' "$ARTIFACT_URL" >> "$GITHUB_STEP_SUMMARY" + fi + printf 'Measurements are informational. Check workload statuses before comparing values.\n' >> "$GITHUB_STEP_SUMMARY" diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 2679fc6d..84f31915 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -75,6 +75,37 @@ This section is for external contributors. PostHog maintainers (members of the P A diff in `public_api_snapshot.txt` (see "CI-aligned checks" above) means your change touches public API. +## SDK performance benchmarks + +Every PR runs the 20-workload [`sdk-benchmarks/v1` specification](https://github.com/PostHog/.github/blob/04f8421ec4daa7f4066eaeb4c53621ba79ddc3ca/.github/workflows/sdk-benchmark-specification.md) against its merge base and exact head. The same head-side fixtures run against separately built, installed gems. Four independent workload shards keep all six alternating base/head pairs on one runner, with a fresh Ruby process for each workload, measurement pass and repetition. No performance percentage is a CI gate; build, correctness and execution errors remain failures. + +The benchmark preparation step installs each checkout's frozen Bundler dependencies (with the repository's RubyGems cooldown), builds its SDK gem, and installs that gem into a fresh external directory. Worker processes use those installed gems, not source imports: + +```bash +python3 -B benchmarks/prepare.py /path/to/base /path/to/build/base +python3 -B benchmarks/prepare.py "$PWD" /path/to/build/head +python3 -B benchmarks/test_benchmarks.py +python3 -B benchmarks/check_native.py /path/to/build/head +python3 -B benchmarks/run.py \ + --base /path/to/base --head "$PWD" --build-directory /path/to/build \ + --output /path/to/results --smoke --repetitions 1 +``` + +Omit `--smoke` and use `--repetitions 6` for full measurements. `--shard` selects `flags-single`, `flags-scaling`, `flags-memory` or `capture`; `--filter` selects a workload prefix for debugging and cannot be combined with a shard. Smoke checks retain the complete inventory with smaller operation counts. Full flags allocation passes use the documented smaller counts recorded in `allocation_operations`. + +Measurement scope: + +- Elapsed time uses `Process.clock_gettime(CLOCK_MONOTONIC)`. CPU uses process time, except enqueue-only workloads use caller-thread time because background serialization has no public enqueue completion signal. +- Allocation passes count Ruby objects with cumulative `GC.stat(:total_allocated_objects)`, including objects later collected and allocations on SDK background threads. Native malloc is excluded. Cumulative allocated bytes are unsupported, not estimated from retained heap. Enqueue-only allocations are also unsupported; completed-delivery workloads measure the full worker pipeline. +- Memory passes sample RSS at the specification's operation checkpoints and before/after work. Linux uses `/proc/self/status`; macOS uses `ps`. Linux lifetime peak RSS uses `VmHWM`; it is explicitly unsupported on macOS. `ObjectSpace.memsize_of_all` reports retained live-object bytes after settling, not total allocator arenas or cumulative allocation volume. Snapshot overhead is part of memory observation. GC runs before/after memory measurement, never inside the workload/checkpoints. +- Eight native threads perform the load scenario under Ruby's GVL; this does not promise eight parallel CPU cores. Load checkpoints join each 32-operation group. +- A coordinator-owned loopback receiver exercises the normal SDK HTTP, batching, byte-size splitting and background-worker paths. Receiver CPU, allocations and retained fixtures are outside the measured Ruby process. Refresh elapsed/allocations include the worker's loopback control request selecting the next generation. No production service or credentials are used. +- Public local-only flag getters and reload APIs are used, with flag-called events disabled. The single-flag getter's deprecation warning occurs during warmup. Capture uses public `batch_size: 100000`, `max_queue_size: 100001`, `flush_interval_seconds: 3600` and `compress_request: false`. Native byte-size batching remains active. Warmup deliveries are verified before counters reset; sequences, payloads, users and final delivery are validated without a rescuing flush for delivery workloads. + +Workers have a 120-second deadline. After two timeouts a coordinator finishes the current pair, saves partial diagnostics and fails. Each CI shard has a 12-minute job and a 10-minute measurement deadline, reserving time for upload. Collection has five minutes and rejects incomplete shard coverage. Run duration must be assessed from hosted CI, not concurrent local measurements. + +The shared reporter validates raw result schema v1 and produces readable tables plus `comparison.json`. Its comment shortlist requires at least a 5% median change and all six pairs agreeing in direction; it is an informational noise filter, not statistical significance. Missing and unsupported counters remain explicit. The read-only producer workflow is separate from the trusted `workflow_run` publisher; automatic PR comments become active only once that caller workflow is on the default branch. + ## Rails package The `posthog-rails` package has its own package-specific guide in [posthog-rails/CONTRIBUTING.md](posthog-rails/CONTRIBUTING.md). diff --git a/benchmarks/.gitignore b/benchmarks/.gitignore new file mode 100644 index 00000000..2b427312 --- /dev/null +++ b/benchmarks/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +results/ diff --git a/benchmarks/check_native.py b/benchmarks/check_native.py new file mode 100644 index 00000000..036f57e2 --- /dev/null +++ b/benchmarks/check_native.py @@ -0,0 +1,97 @@ +"""Run native controls plus regression checks through an installed SDK and loopback receiver.""" + +import argparse +import json +import subprocess +import tempfile +from pathlib import Path + +from fixtures import workloads +from run import HERE, TIMEOUT, WORKER, run_worker, worker_env +from server import BenchmarkServer + + +def main(installation): + with tempfile.TemporaryDirectory(prefix="ruby-controls-") as scratch: + env = worker_env(installation, scratch) + subprocess.run( + ["ruby", str(HERE / "controls.rb")], + env=env, + cwd=scratch, + check=True, + timeout=TIMEOUT, + ) + capture = next( + s for s in workloads(True) if s["id"] == "capture.drain.steady.32" + ) + code = """require ARGV[0] +PostHog::Logging.logger = Logger.new(File::NULL) +w = RubyBenchmark::Workload.new(JSON.parse(ARGV[1]), ARGV[2]) +begin + warmup = w.prepare(200) + raise 'Warmup size changed' unless warmup == 160 + raise 'Warmup was not reset' unless w.control('state')['delivered'].zero? + # Simulate a flush that returns without delivery at the benchmark seam, not inside the SDK. + def w.flush; end + w.run(200) + begin + w.complete + raise 'Completed-delivery verification rescued a missing flush' + rescue RuntimeError => error + raise unless error.message == 'Capture payload, identity or sequence mismatch' + end +ensure + w.close +end +puts 'Incomplete delivery rejected without a rescuing verification flush' +""" + with BenchmarkServer(capture) as server: + subprocess.run( + ["ruby", "-e", code, str(WORKER), json.dumps(capture), server.url], + env=env, + cwd=scratch, + check=True, + timeout=TIMEOUT, + ) + refresh = next(s for s in workloads(True) if s["mode"] == "refresh") + code = """require ARGV[0] +PostHog::Logging.logger = Logger.new(File::NULL) +w = RubyBenchmark::Workload.new(JSON.parse(ARGV[1]), ARGV[2]) +begin + w.prepare(4) + w.reload(1) + w.verify + w.reload(2) + w.verify +ensure + w.close +end +puts 'Changing refresh generations verified for all users' +""" + with BenchmarkServer(refresh) as server: + subprocess.run( + ["ruby", "-e", code, str(WORKER), json.dumps(refresh), server.url], + env=env, + cwd=scratch, + check=True, + timeout=TIMEOUT, + ) + # Real child timeout and capped diagnostic output; no SDK state manipulation. + sleeper = Path(scratch) / "sleeper.rb" + sleeper.write_text("STDERR.write('x' * 100000); STDERR.flush; sleep 60\n") + import run + + original = run.WORKER + try: + run.WORKER = sleeper + raw = run_worker(installation, refresh, "memory", timeout=0.5) + assert raw["status"] == "timeout" and 0 < len(raw["stderr"]) <= 65536 + finally: + run.WORKER = original + print("Real worker timeout classified with capped logs") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("installation", type=Path) + main(parser.parse_args().installation.resolve()) diff --git a/benchmarks/controls.rb b/benchmarks/controls.rb new file mode 100644 index 00000000..4f49c454 --- /dev/null +++ b/benchmarks/controls.rb @@ -0,0 +1,42 @@ +# frozen_string_literal: true + +require_relative 'worker' + +# Native counter and fixture controls run against the installed public SDK entry point. +module RubyBenchmarkControls + def self.check(condition, message) + raise message unless condition + end + + def self.run + users = RubyBenchmark::USERS + check(users.length == 16, 'Expected 16 users') + check(users.map { |u| u[:id] }.uniq.length == 16, 'User identities must be unique') + users.each do |user| + check(RubyBenchmark.expected(0, user), 'Generation zero boolean') + check(!RubyBenchmark.expected(0, user, 1), 'Odd generation boolean') + check(!RubyBenchmark.expected(9, user), 'Inactive flag') + check(!RubyBenchmark.expected(8, user), 'No-match flag') + check(!RubyBenchmark.expected(5, user, 1), 'Odd generation cohort') + check(%w[control test].include?(RubyBenchmark.expected(2, user)), 'Variant') + check(RubyBenchmark.expected(6, user) == RubyBenchmark.expected(7, user), 'Early and late match equivalence') + end + before = GC.stat(:total_allocated_objects) + objects = Array.new(1000) { Object.new } + allocated = GC.stat(:total_allocated_objects) - before + check(allocated >= objects.length, 'Cumulative object counter must include live allocations') + nil + RubyBenchmark.settle + check(GC.stat(:total_allocated_objects) >= before + 1000, 'GC must not reset cumulative allocations') + snapshot = RubyBenchmark.snapshot + check(snapshot[:rss].positive? && snapshot[:heap_used].positive?, 'Native memory counters') + start = Process.clock_gettime(Process::CLOCK_MONOTONIC, :nanosecond) + sleep(0.002) + check(Process.clock_gettime(Process::CLOCK_MONOTONIC, :nanosecond) > start, 'Monotonic clock') + check(RubyBenchmark.describe[:entry] == Gem.loaded_specs.fetch('posthog-ruby').full_gem_path, + 'Installed gem identity') + puts 'Native fixture, allocation, memory, clock and installed-entry controls passed' + end +end + +RubyBenchmarkControls.run if $PROGRAM_NAME == __FILE__ diff --git a/benchmarks/fixtures.py b/benchmarks/fixtures.py new file mode 100644 index 00000000..5c81087d --- /dev/null +++ b/benchmarks/fixtures.py @@ -0,0 +1,216 @@ +"""Deterministic fixtures for sdk-benchmarks/v1.""" + +import hashlib +import json +from typing import Any + +CASES = [ + "boolean", + "rollout", + "variant", + "property", + "regex", + "cohort", + "early", + "late", + "no-match", + "inactive", +] +USERS = [ + { + "distinct_id": f"benchmark-user-{i}", + "person_properties": { + "plan": "free" if i % 2 else "pro", + "email": f"user-{i}@example.test", + "score": i, + }, + } + for i in range(16) +] + + +def prop(key, value, operator="exact"): + return {"key": key, "value": value, "operator": operator, "type": "person"} + + +def group(properties=None, rollout=100): + return {"properties": properties or [], "rollout_percentage": rollout} + + +def definitions(count, generation=0): + flags = [] + for i in range(count): + kind = CASES[i % len(CASES)] + match = group([prop("plan", ["pro"])]) + miss = group([prop("plan", ["never"])]) + groups = [group()] + if kind == "rollout": + groups = [group(rollout=50)] + elif kind == "property": + groups = [match] + elif kind == "regex": + groups = [ + group( + [ + prop( + "email", + f"^(user|refresh-{generation})-[0-9]+@example[.]test$", + "regex", + ) + ] + ) + ] + elif kind == "cohort": + groups = [group([{"key": "id", "value": 1, "type": "cohort"}])] + elif kind == "early": + groups = [match] + [miss] * 7 + elif kind == "late": + groups = [miss] * 7 + [match] + elif kind == "no-match": + groups = [miss] * 8 + filters: dict[str, Any] = {"groups": groups} + if kind == "variant": + filters["multivariate"] = { + "variants": [ + {"key": "control", "rollout_percentage": 50}, + {"key": "test", "rollout_percentage": 50}, + ] + } + flags.append( + { + "id": i + 1, + "key": f"{kind}-{i}", + "name": f"{kind}-{i}", + "active": kind != "inactive" + and (kind != "boolean" or generation % 2 == 0), + "filters": filters, + } + ) + return { + "flags": flags, + "group_type_mapping": {}, + "cohorts": { + "1": { + "type": "AND", + "values": [ + prop("plan", ["pro"]), + prop("score", 100 if generation % 2 else -1, "gt"), + ], + }, + }, + } + + +def bucket(key, distinct_id, salt=""): + return int( + hashlib.sha1(f"{key}.{distinct_id}{salt}".encode()).hexdigest()[:15], 16 + ) / (2**60 - 1) + + +def expected_value(index, user, generation=0): + kind = CASES[index % len(CASES)] + key = f"{kind}-{index}" + if kind == "rollout": + return bucket(key, user["distinct_id"]) < 0.5 + if kind == "variant": + return ( + "control" if bucket(key, user["distinct_id"], "variant") < 0.5 else "test" + ) + if kind == "boolean": + return generation % 2 == 0 + if kind == "cohort" and generation % 2: + return False + if kind in ("property", "cohort", "early", "late"): + return user["person_properties"]["plan"] == "pro" + return kind not in ("inactive", "no-match") + + +def expected_flags(count, user, generation=0): + return { + f"{CASES[i % len(CASES)]}-{i}": expected_value(i, user, generation) + for i in range(count) + } + + +def digest(value): + data = ( + value + if isinstance(value, bytes) + else json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + ) + return hashlib.sha256(data).hexdigest() + + +def workloads(smoke=False): + return [ + *[ + { + "id": f"flags.single.{kind}", + "mode": "single", + "flag_count": 10, + "flag_index": i, + "operations": 160 if smoke else 16000, + "allocation_operations": 160 if smoke else 2000, + "concurrency": 1, + "passes": ["timing", "allocation"], + } + for i, kind in enumerate(CASES[:9]) + ], + *[ + { + "id": f"flags.all.{count}", + "mode": "all", + "flag_count": count, + "operations": 16 if smoke else max(160, 16000 // count), + "allocation_operations": 16 if smoke else max(16, 1600 // count), + "concurrency": 1, + "passes": ["timing", "allocation"], + } + for count in (10, 100, 1000) + ], + { + "id": "flags.load.1000", + "mode": "all", + "flag_count": 1000, + "operations": 32 if smoke else 1600, + "concurrency": 8, + "passes": ["memory"], + }, + { + "id": "flags.refresh.1000", + "mode": "refresh", + "flag_count": 1000, + "operations": 4 if smoke else 40, + "concurrency": 1, + "passes": ["memory"], + }, + *[ + { + "id": f"capture.enqueue.{size}", + "mode": "capture", + "payload_bytes": size, + "operations": 200 if smoke else 1000, + "completion_boundary": "api_return", + "concurrency": 1, + "passes": ["timing"], + } + for size in (32, 4096) + ], + *[ + { + "id": f"capture.drain.{strategy}.{size}", + "mode": "drain", + "drain_strategy": strategy, + "flush_every": ( + 100 if strategy == "steady" else (200 if smoke else 1000) + ), + "completion_boundary": "completed_delivery", + "payload_bytes": size, + "operations": 200 if smoke else 1000, + "concurrency": 1, + "passes": ["timing", "allocation", "memory"], + } + for strategy in ("steady", "burst") + for size in (32, 4096) + ], + ] diff --git a/benchmarks/prepare.py b/benchmarks/prepare.py new file mode 100644 index 00000000..e42ba862 --- /dev/null +++ b/benchmarks/prepare.py @@ -0,0 +1,85 @@ +"""Build and install each SDK gem separately; workers never load a checkout.""" + +import argparse +import json +import os +import subprocess +from pathlib import Path + + +def prepare(directory, output): + directory, output = directory.resolve(), output.resolve() + # An old installed gem with the same version must not mask a new build. + output.mkdir(parents=True, exist_ok=False) + env = { + **os.environ, + "BUNDLE_GEMFILE": str(directory / "Gemfile"), + "BUNDLE_FROZEN": "true", + } + subprocess.run( + ["bundle", "install"], + cwd=directory, + env=env, + check=True, + timeout=120, + ) + dependency = json.loads( + subprocess.check_output( + [ + "bundle", + "exec", + "ruby", + "-rjson", + "-rconcurrent", + "-e", + 'puts JSON.generate({version: Concurrent::VERSION, gem_path: Gem.loaded_specs.fetch("concurrent-ruby").base_dir})', + ], + cwd=directory, + env=env, + text=True, + timeout=120, + ) + ) + artifact = output / "sdk.gem" + subprocess.run( + ["gem", "build", "posthog-ruby.gemspec", "--output", str(artifact)], + cwd=directory, + check=True, + timeout=120, + ) + gems = output / "gems" + subprocess.run( + [ + "gem", + "install", + str(artifact), + "--local", + "--ignore-dependencies", + "--no-document", + "--install-dir", + str(gems), + ], + cwd=output, + check=True, + timeout=120, + ) + specs = list((gems / "specifications").glob("posthog-ruby-*.gemspec")) + if len(specs) != 1: + raise ValueError("Expected one installed SDK gem") + version = specs[0].stem.removeprefix("posthog-ruby-") + info = { + "gem_home": str(gems), + "gem_path": dependency["gem_path"], + "sdk_version": version, + "concurrent_version": dependency["version"], + "artifact": str(artifact), + } + (output / "installation.json").write_text(json.dumps(info, indent=2) + "\n") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("directory", type=Path) + parser.add_argument("output", type=Path) + options = parser.parse_args() + prepare(options.directory, options.output) diff --git a/benchmarks/run.py b/benchmarks/run.py new file mode 100644 index 00000000..c0192d18 --- /dev/null +++ b/benchmarks/run.py @@ -0,0 +1,422 @@ +"""Paired Ruby SDK measurements using head-side fixtures and isolated installed gems.""" + +import argparse +import json +import math +import os +import re +import signal +import subprocess +import sys +import tempfile +import threading +import uuid +from pathlib import Path + +from fixtures import USERS, definitions, digest, workloads +from server import BenchmarkServer + +HERE = Path(__file__).resolve().parent +WORKER = HERE / "worker.rb" +TIMEOUT = 120 +MAX_TIMEOUTS = 2 + + +def git(directory, *args): + return subprocess.check_output( + ["git", "-C", str(directory), *args], text=True + ).strip() + + +def source_digest(): + return digest( + { + p.name: digest(p.read_bytes()) + for p in sorted(HERE.iterdir()) + if p.suffix in (".rb", ".py", ".json") + } + ) + + +def load_shards(): + manifest = json.loads((HERE / "shards.json").read_text()) + if manifest.get("manifest_version") != 1 or manifest.get("sdk") != "posthog-ruby": + raise ValueError("Invalid shard manifest") + shards = {} + seen = set() + for shard in manifest["shards"]: + identity, members = shard["id"], shard["workloads"] + if ( + not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,63}", identity) + or identity in shards + or not members + ): + raise ValueError("Invalid shard identity") + if len(set(members)) != len(members) or seen.intersection(members): + raise ValueError("Duplicate shard workloads") + seen.update(members) + shards[identity] = members + if not 0 < len(shards) <= 16 or any( + seen != {w["id"] for w in workloads(smoke)} for smoke in (True, False) + ): + raise ValueError("Shard manifest must partition full and smoke inventories") + return shards + + +def worker_env(installation, directory): + info = json.loads((installation / "installation.json").read_text()) + env = { + key: value + for key, value in os.environ.items() + if not key.startswith(("BUNDLE_", "RUBY", "GEM_")) + } + for key in ( + "HTTP_PROXY", + "HTTPS_PROXY", + "ALL_PROXY", + "http_proxy", + "https_proxy", + "all_proxy", + "POSTHOG_RELEASE_ID", + ): + env.pop(key, None) + env.update( + GEM_HOME=info["gem_home"], + GEM_PATH=os.pathsep.join([info["gem_home"], info["gem_path"]]), + RUBYOPT="", + BENCHMARK_SDK_VERSION=info["sdk_version"], + BENCHMARK_CONCURRENT_VERSION=info["concurrent_version"], + TMPDIR=str(directory), + HOME=str(directory), + ) + return env + + +def run_worker(installation, spec, measurement_pass, timeout=TIMEOUT): + with tempfile.TemporaryDirectory( + prefix="ruby-benchmark-" + ) as scratch, BenchmarkServer(spec) as server: + scratch = Path(scratch) + input_path, output = scratch / "spec.json", scratch / "result.json" + input_path.write_text(json.dumps({**spec, "pass": measurement_pass})) + process = subprocess.Popen( + ["ruby", str(WORKER), str(input_path), server.url, str(output)], + cwd=scratch, + env=worker_env(installation, scratch), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + start_new_session=True, + ) + buffers = {"stdout": bytearray(), "stderr": bytearray()} + + def read_log(stream, target): + with stream: + while chunk := stream.read(4096): + target.extend(chunk) + del target[:-65536] + + readers = [ + threading.Thread( + target=read_log, args=(getattr(process, name), buffer), daemon=True + ) + for name, buffer in buffers.items() + ] + for reader in readers: + reader.start() + timed_out = False + try: + process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + timed_out = True + os.killpg(process.pid, signal.SIGKILL) + process.wait() + for reader in readers: + reader.join(timeout=2) + logs = { + name: data.decode("utf-8", errors="replace") + for name, data in buffers.items() + } + if process.returncode == 0 and output.exists() and not timed_out: + if output.stat().st_size > 2 * 1024 * 1024: + raise ValueError("Worker result exceeds size limit") + result = json.loads(output.read_text()) + operations = ( + spec.get("allocation_operations", spec["operations"]) + if measurement_pass == "allocation" + else spec["operations"] + ) + if ( + result.get("pass") != measurement_pass + or result.get("operations") != operations + or result.get("status") != "ok" + ): + raise ValueError("Worker result identity differs") + return {**result, **logs} + return { + "status": "timeout" if timed_out else "error", + "pass": measurement_pass, + "code": process.returncode, + "error": ( + "Worker exceeded deadline" + if timed_out + else "Worker did not complete; exit alone does not establish OOM" + ), + **logs, + } + + +def document_for(directory, installation, role, pair_id, suite, shard): + with tempfile.TemporaryDirectory(prefix="ruby-benchmark-describe-") as scratch: + execution = json.loads( + subprocess.check_output( + ["ruby", str(WORKER), "--describe"], + cwd=scratch, + env=worker_env(installation, scratch), + text=True, + timeout=TIMEOUT, + ) + ) + # Fingerprint the installed build, independently of package version. + installed = Path(execution["entry"]) + execution.update( + pair_id=pair_id, + runtime_flags=[], + build_digest=digest( + { + str(p.relative_to(installed)): digest(p.read_bytes()) + for p in sorted((installed / "lib").rglob("*.rb")) + } + ), + dependency_lock_digest=digest((directory / "Gemfile.lock").read_bytes()), + runner_class=( + "github-hosted-or-self-hosted" if os.getenv("RUNNER_NAME") else "local" + ), + runner_image=os.getenv("ImageVersion"), + resource_limits={ + "worker_timeout_ms": TIMEOUT * 1000, + "max_worker_timeouts": MAX_TIMEOUTS, + "process_rss_limit_bytes": None, + }, + ) + if shard: + execution["shard_id"] = shard + return { + "schema_version": 1, + "sdk": "posthog-ruby", + "revision": { + "role": role, + "commit": git(directory, "rev-parse", "HEAD"), + "dirty": bool(git(directory, "status", "--porcelain")), + }, + "suite": suite, + "execution": execution, + "benchmarks": [], + } + + +def append_sample(record, raw, repetition): + record["runs"].append({"repetition": repetition, **raw}) + if raw["status"] != "ok": + record["status"] = raw["status"] + return + + def add(name, unit, method, value): + if not math.isfinite(value) or value < 0: + raise ValueError(f"Invalid metric: {name}") + metric = next((m for m in record["metrics"] if m["id"] == name), None) + if metric is None: + metric = { + "id": name, + "unit": unit, + "method": method, + "quality": "measured", + "direction": "lower", + "status": "ok", + "samples": [], + } + record["metrics"].append(metric) + metric["samples"].append( + {"repetition": repetition, "value": value, "operations": raw["operations"]} + ) + + def unsupported(name, unit, reason): + if not any(m["id"] == name for m in record["metrics"]): + record["metrics"].append( + {"id": name, "unit": unit, "status": "unsupported", "reason": reason} + ) + + if raw["pass"] == "timing": + add( + "elapsed_time_per_op", + "ns/op", + "Process.clock_gettime.CLOCK_MONOTONIC", + raw["elapsed_time_per_op"], + ) + enqueue = record["parameters"]["mode"] == "capture" + add( + "caller_thread_cpu_time_per_op" if enqueue else "process_cpu_time_per_op", + "ns/op", + ( + "Process.clock_gettime.CLOCK_THREAD_CPUTIME_ID" + if enqueue + else "Process.clock_gettime.CLOCK_PROCESS_CPUTIME_ID" + ), + raw["cpu_time_per_op"], + ) + if enqueue: + for name, unit in ( + ("allocated_bytes_per_op", "bytes/op"), + ("allocations_per_op", "count/op"), + ): + unsupported( + name, + unit, + "Background serialization has no public enqueue completion boundary; use completed-delivery workloads for process-wide allocation counts.", + ) + elif raw["pass"] == "allocation": + add( + "allocations_per_op", + "count/op", + "GC.stat.total_allocated_objects; all worker threads; Ruby objects excluding native malloc", + raw["allocated_objects"] / raw["operations"], + ) + unsupported( + "allocated_bytes_per_op", + "bytes/op", + "Ruby has no cumulative allocated-byte counter here; retained ObjectSpace bytes are not allocation volume.", + ) + elif raw["pass"] == "memory": + for name, value in ( + ("idle_process_rss", raw["before"]["rss"]), + ("observed_workload_peak_rss", raw["observed_peak_rss"]), + ("post_settling_process_rss", raw["after_settling"]["rss"]), + ): + add( + name, + "bytes", + "proc.status.VmRSS or ps.rss; operation checkpoints", + value, + ) + add( + "post_settling_heap_used", + "bytes", + "ObjectSpace.memsize_of_all; retained live objects, not allocator arena size", + raw["after_settling"]["heap_used"], + ) + if raw["process_lifetime_peak_rss"] is not None: + add( + "process_lifetime_peak_rss", + "bytes", + "proc.status.VmHWM; includes setup and warmup", + raw["process_lifetime_peak_rss"], + ) + else: + unsupported( + "process_lifetime_peak_rss", + "bytes", + "No process-lifetime RSS high-water counter on this platform; not inferred from checkpoints.", + ) + + +def main(args=None): + parser = argparse.ArgumentParser(description=__doc__) + for name in ("base", "head", "build_directory", "output"): + parser.add_argument("--" + name.replace("_", "-"), type=Path, required=True) + parser.add_argument("--smoke", action="store_true") + parser.add_argument("--repetitions", type=int, default=6, choices=range(1, 31)) + selection = parser.add_mutually_exclusive_group() + selection.add_argument("--shard") + selection.add_argument("--filter", default="") + options = parser.parse_args(args) + shards = load_shards() + if options.shard and options.shard not in shards: + parser.error("Unknown shard") + specs = ( + [s for s in workloads(options.smoke) if s["id"] in shards[options.shard]] + if options.shard + else [s for s in workloads(options.smoke) if s["id"].startswith(options.filter)] + ) + if not specs: + parser.error("No matching workloads") + suite = { + "id": "ruby-sdk-performance", + "version": 2, + "specification": "sdk-benchmarks/v1", + "source_commit": git(HERE, "rev-parse", "HEAD"), + "source_digest": source_digest(), + "profile": "smoke" if options.smoke else "full", + "repetitions": options.repetitions, + "transport": "coordinator-owned loopback HTTP receiver; uncompressed; native batch byte limits retained", + "capture": { + "batch_size": 100000, + "max_queue_size": 100001, + "flush_interval_seconds": 3600, + }, + "warmup": "two refreshes; otherwise min(operations, 32 for 1000 flags or 160); verified reset and generation zero restored", + "settling": "two GC.start calls separated by Thread.pass; no forced GC in measured work", + "memory_sampling": "after every 32 load operations, each refresh, each 100 captures, and final work; ObjectSpace and RSS snapshots", + "allocation_scope": "GC.stat Ruby object allocations across worker threads; excludes coordinator receiver and native malloc", + } + pair_id = str(uuid.uuid4()) + directories = {"base": options.base.resolve(), "head": options.head.resolve()} + builds = {role: options.build_directory.resolve() / role for role in directories} + documents = { + role: document_for(directory, builds[role], role, pair_id, suite, options.shard) + for role, directory in directories.items() + } + for key in ("runtime", "runtime_version", "measurement_tools", "os", "arch"): + if documents["base"]["execution"][key] != documents["head"]["execution"][key]: + raise ValueError(f"Base/head {key} differ") + options.output.mkdir(parents=True, exist_ok=True) + + def save(): + for role, document in documents.items(): + temporary = options.output / f"{role}.json.tmp" + temporary.write_text(json.dumps(document, allow_nan=False, indent=2) + "\n") + temporary.replace(options.output / f"{role}.json") + + failures = timeouts = 0 + for spec in specs: + records = {} + for role in directories: + record = { + "id": spec["id"], + "parameters": spec, + "fixture_digest": digest( + { + "spec": spec, + "users": USERS, + "definitions": ( + definitions(spec["flag_count"]) + if "flag_count" in spec + else None + ), + } + ), + "status": "ok", + "metrics": [], + "runs": [], + } + records[role] = record + documents[role]["benchmarks"].append(record) + for repetition in range(options.repetitions): + for measurement_pass in spec["passes"]: + for role in ("head", "base") if repetition % 2 else ("base", "head"): + raw = run_worker(builds[role], spec, measurement_pass) + append_sample(records[role], raw, repetition) + failures += raw["status"] != "ok" + timeouts += raw["status"] == "timeout" + print( + f'{spec["id"]} {measurement_pass} {repetition + 1}/{options.repetitions} {role}: {raw["status"]}', + flush=True, + ) + save() + if timeouts >= MAX_TIMEOUTS: + return 1 + if any(record["status"] != "ok" for record in records.values()): + break + return int(failures > 0) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarks/server.py b/benchmarks/server.py new file mode 100644 index 00000000..b119c2ff --- /dev/null +++ b/benchmarks/server.py @@ -0,0 +1,159 @@ +"""Bounded loopback receiver owned by the coordinator, never the measured worker. + +The SDK has no public transport injection option. HTTP preserves its normal +loading, serialization, background-consumer and flush paths. +""" + +import json +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any +from urllib.parse import urlsplit + +from fixtures import USERS, definitions + + +class BenchmarkServer: + def __init__(self, spec, batch_delay=0): + self.spec = spec + self.lock = threading.Lock() + self.payload = "x" * spec.get("payload_bytes", 0) + self.body = json.dumps(definitions(spec.get("flag_count", 0))).encode() + self.seen = bytearray(spec["operations"]) + self.expected_sequences = 0 + self.state: dict[str, Any] = { + "definitions": 0, + "delivered": 0, + "requests": 0, + "invalid_events": 0, + "unique_sequences": 0, + "missing_sequences": 0, + "delivered_by_user": {user["distinct_id"]: 0 for user in USERS}, + "unexpected": [], + } + owner = self + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def reply(self, status, body): + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self): + with owner.lock: + path = urlsplit(self.path).path.rstrip("/") + if path == "/flags/definitions": + owner.state["definitions"] += 1 + self.reply(200, owner.body) + elif path == "/__benchmark/state": + self.reply(200, json.dumps(owner.state).encode()) + else: + self.unexpected(path) + + def unexpected(self, path): + if len(owner.state["unexpected"]) < 20: + owner.state["unexpected"].append(path) + self.reply(404, b'{"error":"unexpected benchmark request"}') + + def do_POST(self): + with owner.lock: + path = urlsplit(self.path).path.rstrip("/") + try: + size = int(self.headers.get("Content-Length", "0")) + if not 0 < size <= 20 * 1024 * 1024: + raise ValueError("Invalid body length") + body = json.loads(self.rfile.read(size)) + if path == "/__benchmark/reset": + operations = body["operations"] + if ( + type(operations) is not int + or not 0 <= operations <= spec["operations"] + ): + raise ValueError("Invalid phase operations") + owner.seen[:] = bytes(len(owner.seen)) + owner.expected_sequences = operations + owner.state.update( + definitions=0, + delivered=0, + requests=0, + invalid_events=0, + unique_sequences=0, + missing_sequences=operations, + delivered_by_user={ + user["distinct_id"]: 0 for user in USERS + }, + unexpected=[], + ) + self.reply(200, b"{}") + elif path == "/__benchmark/definitions": + generation = body["generation"] + if ( + not isinstance(generation, int) + or not 0 <= generation <= 1000 + ): + raise ValueError("Invalid generation") + owner.body = json.dumps( + definitions(spec["flag_count"], generation) + ).encode() + self.reply(200, b"{}") + elif path == "/batch": + if batch_delay: + time.sleep(batch_delay) + batch = body["batch"] + if not isinstance(batch, list): + raise ValueError("Invalid batch") + owner.state["requests"] += 1 + owner.state["delivered"] += len(batch) + for event in batch: + properties = event.get("properties", {}) + user = event.get("distinct_id") + sequence = properties.get("sequence") + valid_sequence = ( + type(sequence) is int + and 0 <= sequence < owner.expected_sequences + ) + if ( + event.get("event") != "benchmark" + or user not in owner.state["delivered_by_user"] + or properties.get("payload") != owner.payload + or properties.get("plan") != "pro" + or not valid_sequence + or user + != USERS[sequence % len(USERS)]["distinct_id"] + ): + owner.state["invalid_events"] += 1 + else: + owner.state["delivered_by_user"][user] += 1 + if owner.seen[sequence]: + owner.state["invalid_events"] += 1 + else: + owner.seen[sequence] = 1 + owner.state["unique_sequences"] += 1 + owner.state["missing_sequences"] -= 1 + self.reply(200, b'{"status":1}') + else: + self.unexpected(path) + except (ValueError, KeyError, TypeError, AttributeError): + owner.state["invalid_events"] += 1 + self.reply(400, b'{"error":"invalid benchmark body"}') + + self.http = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.thread = threading.Thread( + target=self.http.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True + ) + self.url = f"http://127.0.0.1:{self.http.server_port}" + + def __enter__(self): + self.thread.start() + return self + + def __exit__(self, *_): + self.http.shutdown() + self.http.server_close() + self.thread.join() diff --git a/benchmarks/shards.json b/benchmarks/shards.json new file mode 100644 index 00000000..8a17e284 --- /dev/null +++ b/benchmarks/shards.json @@ -0,0 +1,39 @@ +{ + "manifest_version": 1, + "sdk": "posthog-ruby", + "shards": [ + { + "id": "flags-single", + "workloads": [ + "flags.single.boolean", + "flags.single.rollout", + "flags.single.variant", + "flags.single.property", + "flags.single.regex", + "flags.single.cohort", + "flags.single.early", + "flags.single.late", + "flags.single.no-match" + ] + }, + { + "id": "flags-scaling", + "workloads": ["flags.all.10", "flags.all.100", "flags.all.1000"] + }, + { + "id": "flags-memory", + "workloads": ["flags.load.1000", "flags.refresh.1000"] + }, + { + "id": "capture", + "workloads": [ + "capture.enqueue.32", + "capture.enqueue.4096", + "capture.drain.steady.32", + "capture.drain.steady.4096", + "capture.drain.burst.32", + "capture.drain.burst.4096" + ] + } + ] +} diff --git a/benchmarks/test_benchmarks.py b/benchmarks/test_benchmarks.py new file mode 100644 index 00000000..0ab29faa --- /dev/null +++ b/benchmarks/test_benchmarks.py @@ -0,0 +1,343 @@ +"""Coordinator, fixture and real public-entry controls. No external service required.""" + +import copy +import json +import subprocess +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch +from urllib.request import Request, urlopen + +import fixtures +import run +from server import BenchmarkServer + +HERE = Path(__file__).resolve().parent + + +class Controls(unittest.TestCase): + def test_inventory_and_partition(self): + shards = run.load_shards() + self.assertEqual(len(shards), 4) + for smoke in (True, False): + specs = fixtures.workloads(smoke) + self.assertEqual(len(specs), 20) + self.assertEqual(len({s["id"] for s in specs}), 20) + self.assertEqual( + set().union(*(set(w) for w in shards.values())), + {s["id"] for s in specs}, + ) + self.assertEqual(len([s for s in specs if s["mode"] == "single"]), 9) + for spec in specs: + ( + self.assertEqual(spec["operations"], 200 if smoke else 1000) + if "payload_bytes" in spec + else None + ) + if spec["mode"] == "drain": + self.assertEqual( + spec["flush_every"], + ( + 100 + if spec["drain_strategy"] == "steady" + else spec["operations"] + ), + ) + self.assertEqual(spec["passes"], ["timing", "allocation", "memory"]) + + def test_flags_fixtures(self): + doc = fixtures.definitions(1000) + self.assertEqual(len(doc["flags"]), 1000) + self.assertEqual(len(fixtures.USERS), 16) + self.assertEqual(doc["flags"][9]["active"], False) + self.assertEqual(len(doc["flags"][6]["filters"]["groups"]), 8) + self.assertEqual(len(doc["flags"][7]["filters"]["groups"]), 8) + self.assertNotEqual( + fixtures.definitions(1000, 0), fixtures.definitions(1000, 1) + ) + self.assertEqual( + fixtures.definitions(1000, 0), + fixtures.definitions(1000, 2) + | {"flags": fixtures.definitions(1000, 0)["flags"]}, + ) + + def test_source_identity_includes_manifest(self): + before = run.source_digest() + original = (HERE / "shards.json").read_bytes() + real = Path.read_bytes + with patch.object( + Path, + "read_bytes", + lambda p: original + b" " if p.name == "shards.json" else real(p), + ): + self.assertNotEqual(run.source_digest(), before) + + def test_bad_manifests_rejected(self): + manifest = json.loads((HERE / "shards.json").read_text()) + for mutate in ( + lambda m: m["shards"][0]["workloads"].pop(), + lambda m: m["shards"][1]["workloads"].append( + m["shards"][0]["workloads"][0] + ), + lambda m: m["shards"][0].update(id="../unsafe"), + ): + changed = copy.deepcopy(manifest) + mutate(changed) + with patch.object( + Path, "read_text", return_value=json.dumps(changed) + ), self.assertRaises(ValueError): + run.load_shards() + + def test_metrics_keep_failures_and_unsupported_distinct(self): + record = { + "status": "ok", + "parameters": {"mode": "all"}, + "runs": [], + "metrics": [], + } + run.append_sample( + record, + { + "status": "ok", + "pass": "allocation", + "operations": 10, + "allocated_objects": 100, + }, + 0, + ) + values = {m["id"]: m for m in record["metrics"]} + self.assertEqual(values["allocations_per_op"]["samples"][0]["value"], 10) + self.assertEqual(values["allocated_bytes_per_op"]["status"], "unsupported") + run.append_sample(record, {"status": "timeout", "pass": "allocation"}, 1) + self.assertEqual(record["status"], "timeout") + self.assertEqual(len(values["allocations_per_op"]["samples"]), 1) + + def test_invalid_counter_rejected(self): + record = { + "status": "ok", + "parameters": {"mode": "all"}, + "runs": [], + "metrics": [], + } + with self.assertRaises(ValueError): + run.append_sample( + record, + { + "status": "ok", + "pass": "allocation", + "operations": 1, + "allocated_objects": -1, + }, + 0, + ) + + def test_enqueue_uses_caller_thread_cpu(self): + record = { + "status": "ok", + "parameters": {"mode": "capture"}, + "runs": [], + "metrics": [], + } + run.append_sample( + record, + { + "status": "ok", + "pass": "timing", + "operations": 1, + "elapsed_time_per_op": 2, + "cpu_time_per_op": 1, + }, + 0, + ) + self.assertEqual( + {m["id"] for m in record["metrics"]}, + { + "elapsed_time_per_op", + "caller_thread_cpu_time_per_op", + "allocations_per_op", + "allocated_bytes_per_op", + }, + ) + + def test_timeout_finishes_pair_and_stops(self): + doc = { + "schema_version": 1, + "execution": { + k: "same" + for k in ( + "runtime", + "runtime_version", + "measurement_tools", + "os", + "arch", + ) + }, + "benchmarks": [], + } + calls = [] + + def fake_worker(installation, spec, measurement_pass): + calls.append(installation.name) + return {"status": "timeout", "pass": measurement_pass} + + with tempfile.TemporaryDirectory() as t, patch.object( + run, "document_for", side_effect=lambda *a: copy.deepcopy(doc) + ), patch.object(run, "run_worker", side_effect=fake_worker): + code = run.main( + [ + "--base", + t, + "--head", + t, + "--build-directory", + t, + "--output", + t, + "--shard", + "flags-single", + ] + ) + self.assertEqual(code, 1) + self.assertEqual(calls, ["base", "head"]) + saved = json.loads((Path(t) / "head.json").read_text()) + self.assertEqual(saved["benchmarks"][0]["status"], "timeout") + + def test_alternating_pairs(self): + doc = { + "execution": { + k: "same" + for k in ( + "runtime", + "runtime_version", + "measurement_tools", + "os", + "arch", + ) + }, + "benchmarks": [], + } + calls = [] + + def worker(installation, spec, measurement_pass): + calls.append(installation.name) + return { + "status": "ok", + "pass": measurement_pass, + "operations": 1, + "elapsed_time_per_op": 1, + "cpu_time_per_op": 1, + "allocated_objects": 1, + } + + with tempfile.TemporaryDirectory() as t, patch.object( + run, "document_for", side_effect=lambda *a: copy.deepcopy(doc) + ), patch.object(run, "run_worker", side_effect=worker): + self.assertEqual( + run.main( + [ + "--base", + t, + "--head", + t, + "--build-directory", + t, + "--output", + t, + "--filter", + "flags.single.boolean", + "--repetitions", + "2", + ] + ), + 0, + ) + self.assertEqual( + calls, ["base", "head", "base", "head", "head", "base", "head", "base"] + ) + + def test_describe_is_bounded(self): + with tempfile.TemporaryDirectory() as t, patch.object( + run, "worker_env", return_value={} + ), patch.object( + run.subprocess, + "check_output", + side_effect=subprocess.TimeoutExpired("ruby", run.TIMEOUT), + ) as command: + with self.assertRaises(subprocess.TimeoutExpired): + run.document_for(Path(t), Path(t), "head", "pair", {}, None) + self.assertEqual(command.call_args.kwargs["timeout"], run.TIMEOUT) + + def test_unknown_shard_and_filter_conflict(self): + with self.assertRaises(SystemExit): + run.main( + [ + "--base", + ".", + "--head", + ".", + "--build-directory", + ".", + "--output", + ".", + "--shard", + "unknown", + ] + ) + with self.assertRaises(SystemExit): + run.main( + [ + "--base", + ".", + "--head", + ".", + "--build-directory", + ".", + "--output", + ".", + "--shard", + "capture", + "--filter", + "flags", + ] + ) + + def test_receiver_sequence_validation(self): + spec = next( + s for s in fixtures.workloads(True) if s["id"] == "capture.drain.burst.32" + ) + + def request(server, path, body=None): + req = Request( + server.url + path, + data=json.dumps(body).encode() if body is not None else None, + ) + with urlopen(req, timeout=5) as response: + return json.load(response) + + def event(i): + return { + "event": "benchmark", + "distinct_id": fixtures.USERS[i % 16]["distinct_id"], + "properties": {"sequence": i, "plan": "pro", "payload": "x" * 32}, + } + + with BenchmarkServer(spec) as server: + request(server, "/__benchmark/reset", {"operations": 32}) + request( + server, "/batch", {"batch": [event(i) for i in reversed(range(32))]} + ) + state = request(server, "/__benchmark/state") + self.assertEqual(state["missing_sequences"], 0) + self.assertEqual(state["invalid_events"], 0) + request(server, "/__benchmark/reset", {"operations": 32}) + # Balanced duplicate/loss preserves event/user totals but must fail. + request(server, "/batch", {"batch": [event(i) for i in range(16)] * 2}) + state = request(server, "/__benchmark/state") + self.assertEqual(state["delivered"], 32) + self.assertGreater(state["missing_sequences"], 0) + self.assertGreater(state["invalid_events"], 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/benchmarks/worker.rb b/benchmarks/worker.rb new file mode 100644 index 00000000..ec2c2be9 --- /dev/null +++ b/benchmarks/worker.rb @@ -0,0 +1,291 @@ +# frozen_string_literal: true + +require 'json' +require 'net/http' +require 'digest' +require 'objspace' +require 'logger' +require 'rbconfig' + +gem 'concurrent-ruby', ENV.fetch('BENCHMARK_CONCURRENT_VERSION') +gem 'posthog-ruby', ENV.fetch('BENCHMARK_SDK_VERSION') +require 'posthog' + +module RubyBenchmark + CASES = %w[boolean rollout variant property regex cohort early late no-match inactive].freeze + USERS = Array.new(16) do |i| + { id: "benchmark-user-#{i}", properties: { plan: i.even? ? 'pro' : 'free', + email: "user-#{i}@example.test", score: i } } + end.freeze + + def self.require_condition(condition, message) + raise message unless condition + end + + def self.expected(index, user, generation = 0) + kind = CASES[index % 10] + return generation.even? if kind == 'boolean' + return false if %w[inactive no-match].include?(kind) || (kind == 'cohort' && generation.odd?) + return user[:properties][:plan] == 'pro' if %w[property cohort early late].include?(kind) + + if %w[rollout variant].include?(kind) + salt = kind == 'variant' ? 'variant' : '' + bucket = Digest::SHA1.hexdigest("#{kind}-#{index}.#{user[:id]}#{salt}")[0, 15].to_i(16).to_f / ((2**60) - 1) + return bucket < 0.5 if kind == 'rollout' + + return bucket < 0.5 ? 'control' : 'test' + end + true + end + + def self.settle + GC.start + Thread.pass + GC.start + end + + def self.snapshot + # ObjectSpace reports retained live-object bytes, not cumulative allocation volume. + heap = ObjectSpace.memsize_of_all + rss = if File.file?('/proc/self/status') + File.read('/proc/self/status')[/^VmRSS:\s+(\d+)/, 1].to_i * 1024 + else + IO.popen(['ps', '-o', 'rss=', '-p', Process.pid.to_s], &:read).to_i * 1024 + end + require_condition(rss.positive?, 'RSS counter unavailable') + { rss: rss, heap_used: heap } + end + + def self.describe + { + runtime: RUBY_ENGINE, runtime_version: RUBY_VERSION, os: RbConfig::CONFIG['host_os'], + arch: RbConfig::CONFIG['host_cpu'], + cpu: if File.file?('/proc/cpuinfo') + File.read('/proc/cpuinfo')[/^model name\s*:\s*(.+)/, 1] || 'unknown' + else + IO.popen(['sysctl', '-n', 'machdep.cpu.brand_string'], &:read).strip + end, + kernel: IO.popen(['uname', '-r'], &:read).strip, + sdk_version: PostHog::VERSION, + entry: Gem.loaded_specs.fetch('posthog-ruby').full_gem_path, + packages: { 'concurrent-ruby' => Concurrent::VERSION }, + measurement_tools: { ruby: RUBY_DESCRIPTION, objectspace: 'stdlib' } + } + end + + class Workload + def initialize(spec, url) + @spec = spec + @url = url + @generation = 0 + @captured = 0 + @errors = [] + @client = PostHog::Client.new( + api_key: 'benchmark-project', secret_key: spec.key?('flag_count') ? 'benchmark-secret' : nil, + host: url, batch_size: 100_000, max_queue_size: 100_001, flush_interval_seconds: 3600, + feature_flags_polling_interval: 3600, compress_request: false, max_retries: 0, + feature_flag_request_max_retries: 0, disable_singleton_warning: true, + on_error: ->(status, error) { @errors << [status, error.to_s[0, 300]] if @errors.length < 10 } + ) + end + + def control(path, value = nil) + uri = URI("#{@url}/__benchmark/#{path}") + http = Net::HTTP.new(uri.host, uri.port, nil) + http.open_timeout = http.read_timeout = 5 + request = value ? Net::HTTP::Post.new(uri) : Net::HTTP::Get.new(uri) + request.body = JSON.generate(value) if value + response = http.request(request) + RubyBenchmark.require_condition(response.code == '200', 'Benchmark control failed') + JSON.parse(response.body) + end + + def reload(generation) + control('definitions', { generation: generation }) + @client.reload_feature_flags + @generation = generation + end + + def all_flags(user) + @client.get_all_flags(user[:id], person_properties: user[:properties], only_evaluate_locally: true) + end + + def single_flag(user) + index = @spec.fetch('flag_index') + @client.get_feature_flag( + "#{CASES[index]}-#{index}", user[:id], person_properties: user[:properties], + only_evaluate_locally: true, send_feature_flag_events: false + ) + end + + def prepare_events(operations) + @events = Array.new(operations) do |i| + { event: 'benchmark', distinct_id: USERS[i % 16][:id], + properties: { payload: 'x' * @spec.fetch('payload_bytes'), plan: 'pro', sequence: i } } + end + end + + def run(operations) + mode = @spec.fetch('mode') + if mode == 'refresh' + operations.times do |i| + reload(i + 1) + yield(i + 1) if block_given? + end + elsif %w[capture drain].include?(mode) + operations.times do |i| + RubyBenchmark.require_condition(@client.capture(@events[i]), 'Capture rejected') + @captured += 1 + flush if mode == 'drain' && @spec['drain_strategy'] == 'steady' && + (((i + 1) % @spec['flush_every']).zero? || i + 1 == operations) + yield(i + 1) if block_given? && ((i + 1) % 100).zero? + end + flush if mode == 'drain' && @spec['drain_strategy'] == 'burst' + elsif @spec.fetch('concurrency') > 1 + # Join in deterministic 32-operation groups, preserving native GVL-constrained threads. + operations.times.each_slice(32) do |positions| + threads = positions.each_slice(4).map do |slice| + Thread.new { slice.each { |i| all_flags(USERS[i % 16]) } } + end + threads.each(&:value) + yield(positions.last + 1) if block_given? + end + else + operations.times do |i| + mode == 'single' ? single_flag(USERS[i % 16]) : all_flags(USERS[i % 16]) + end + end + end + + def flush + RubyBenchmark.require_condition(@client.flush(timeout: 10), 'Public flush timed out') + end + + def verify + RubyBenchmark.require_condition(@errors.empty?, "SDK callback errors: #{@errors.inspect}") + state = control('state') + RubyBenchmark.require_condition(state.fetch('unexpected').empty?, 'Unexpected network request') + if @spec.key?('flag_count') + RubyBenchmark.require_condition(@client.feature_flags_loaded?, 'Flag definitions not loaded') + USERS.each do |user| + expected = @spec['flag_count'].times.to_h do |i| + ["#{CASES[i % 10]}-#{i}", RubyBenchmark.expected(i, user, @generation)] + end + actual = all_flags(user) + RubyBenchmark.require_condition(actual == expected, 'All-flags correctness mismatch') + next unless @spec['mode'] == 'single' + + i = @spec.fetch('flag_index') + RubyBenchmark.require_condition(single_flag(user) == RubyBenchmark.expected(i, user, @generation), + 'Single-flag correctness mismatch') + end + if @spec['mode'] == 'refresh' + RubyBenchmark.require_condition(state.fetch('definitions') == @generation, + 'Reload did not request definitions') + end + else + expected_counts = USERS.each_with_index.to_h do |user, i| + [user[:id], (@captured / 16) + (i < @captured % 16 ? 1 : 0)] + end + valid = state['delivered'] == @captured && state['invalid_events'].zero? && + state['unique_sequences'] == @captured && state['missing_sequences'].zero? && + state['delivered_by_user'] == expected_counts + RubyBenchmark.require_condition(valid, 'Capture payload, identity or sequence mismatch') + end + end + + def prepare(operations) + warmup = if @spec['mode'] == 'refresh' + 2 + else + [operations, @spec['flag_count'] == 1000 ? 32 : 160].min + end + control('reset', { operations: @spec.key?('payload_bytes') ? warmup : 0 }) + prepare_events(warmup) if @spec.key?('payload_bytes') + run(warmup) + flush if @spec['mode'] == 'capture' + verify + reload(0) if @spec.key?('flag_count') + control('reset', { operations: @spec.key?('payload_bytes') ? operations : 0 }) + @captured = 0 + prepare_events(operations) if @spec.key?('payload_bytes') + warmup + end + + def complete + # Enqueue-only verification flush is outside measurement. Delivery verification never rescues a flush. + flush if @spec['mode'] == 'capture' + verify + end + + def close + RubyBenchmark.require_condition(@client.shutdown(timeout: 10), 'Public shutdown timed out') + end + end + + def self.measure(spec, url) + operations = if spec['pass'] == 'allocation' + spec.fetch('allocation_operations', + spec['operations']) + else + spec['operations'] + end + workload = Workload.new(spec, url) + begin + warmup = workload.prepare(operations) + settle + result = { status: 'ok', pass: spec['pass'], operations: operations, warmup_operations: warmup } + case spec['pass'] + when 'timing' + cpu_clock = spec['mode'] == 'capture' ? Process::CLOCK_THREAD_CPUTIME_ID : Process::CLOCK_PROCESS_CPUTIME_ID + cpu = Process.clock_gettime(cpu_clock, :nanosecond) + start = Process.clock_gettime(Process::CLOCK_MONOTONIC, :nanosecond) + workload.run(operations) + elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC, :nanosecond) - start + cpu = Process.clock_gettime(cpu_clock, :nanosecond) - cpu + result.merge!(elapsed_ns: elapsed, elapsed_time_per_op: elapsed.to_f / operations, + cpu_time_per_op: cpu.to_f / operations) + when 'allocation' + before = GC.stat(:total_allocated_objects) + workload.run(operations) + result[:allocated_objects] = GC.stat(:total_allocated_objects) - before + when 'memory' + before = snapshot + checkpoints = [] + workload.run(operations) { |i| checkpoints << { operations: i, **snapshot } } + if checkpoints.empty? || checkpoints.last[:operations] != operations + checkpoints << { operations: operations, + **snapshot } + end + after = snapshot + settle + after_settling = snapshot + high_water = if File.file?('/proc/self/status') + File.read('/proc/self/status')[/^VmHWM:\s+(\d+)/, + 1].to_i * 1024 + end + result.merge!(before: before, after: after, after_settling: after_settling, checkpoints: checkpoints, + observed_peak_rss: [before[:rss], after[:rss], *checkpoints.map { |p| p[:rss] }].max, + process_lifetime_peak_rss: high_water) + else + raise 'Unknown measurement pass' + end + workload.complete + result + ensure + workload.close + end + end +end + +if $PROGRAM_NAME == __FILE__ + PostHog::Logging.logger = Logger.new(File::NULL) + if ARGV == ['--describe'] + puts JSON.generate(RubyBenchmark.describe) + else + spec_path, url, output = ARGV + spec = JSON.parse(File.read(spec_path)) + result = RubyBenchmark.measure(spec, url) + File.write(output, JSON.generate(result)) + end +end From 4933353ad445bc09a308f79bef9cbfb6f81712b7 Mon Sep 17 00:00:00 2001 From: Dustin Byrne Date: Tue, 3 Nov 2026 14:46:51 -0500 Subject: [PATCH 2/3] ci: pin SDK benchmark reporting to mainline --- .github/workflows/benchmark-report.yml | 2 +- .github/workflows/benchmarks.yml | 2 +- CONTRIBUTING.md | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/benchmark-report.yml b/.github/workflows/benchmark-report.yml index 6de61dc9..59abcbc9 100644 --- a/.github/workflows/benchmark-report.yml +++ b/.github/workflows/benchmark-report.yml @@ -13,7 +13,7 @@ jobs: contents: read actions: read pull-requests: write - uses: PostHog/.github/.github/workflows/sdk-benchmark-report.yml@2a6111d08d784c0ca84f7f97f32a13fdd6d31449 # SDK benchmark workflow (PostHog/.github PR #84) + uses: PostHog/.github/.github/workflows/sdk-benchmark-report.yml@3322f26f21f088ad5e5960d65a916971e309ac0b # SDK benchmark workflow (PostHog/.github PR #84) with: workflow-path: .github/workflows/benchmarks.yml artifact-name: ruby-benchmark-results diff --git a/.github/workflows/benchmarks.yml b/.github/workflows/benchmarks.yml index 061dedfc..f746c064 100644 --- a/.github/workflows/benchmarks.yml +++ b/.github/workflows/benchmarks.yml @@ -106,7 +106,7 @@ jobs: merge-multiple: false - name: Collect and render benchmark comparison - uses: PostHog/.github/.github/actions/benchmark-report@04f8421ec4daa7f4066eaeb4c53621ba79ddc3ca # SDK benchmark reporting (PostHog/.github PR #84) + uses: PostHog/.github/.github/actions/benchmark-report@3322f26f21f088ad5e5960d65a916971e309ac0b # SDK benchmark reporting (PostHog/.github PR #84) with: mode: collect source-status: ${{ needs.benchmark.result }} diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 84f31915..1efa9631 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -77,7 +77,7 @@ A diff in `public_api_snapshot.txt` (see "CI-aligned checks" above) means your c ## SDK performance benchmarks -Every PR runs the 20-workload [`sdk-benchmarks/v1` specification](https://github.com/PostHog/.github/blob/04f8421ec4daa7f4066eaeb4c53621ba79ddc3ca/.github/workflows/sdk-benchmark-specification.md) against its merge base and exact head. The same head-side fixtures run against separately built, installed gems. Four independent workload shards keep all six alternating base/head pairs on one runner, with a fresh Ruby process for each workload, measurement pass and repetition. No performance percentage is a CI gate; build, correctness and execution errors remain failures. +Every PR runs the 20-workload [`sdk-benchmarks/v1` specification](https://github.com/PostHog/.github/blob/3322f26f21f088ad5e5960d65a916971e309ac0b/.github/workflows/sdk-benchmark-specification.md) against its merge base and exact head. The same head-side fixtures run against separately built, installed gems. Four independent workload shards keep all six alternating base/head pairs on one runner, with a fresh Ruby process for each workload, measurement pass and repetition. No performance percentage is a CI gate; build, correctness and execution errors remain failures. The benchmark preparation step installs each checkout's frozen Bundler dependencies (with the repository's RubyGems cooldown), builds its SDK gem, and installs that gem into a fresh external directory. Worker processes use those installed gems, not source imports: From 1b06f3c88322b89fb36182f67d225ddd003f3e2d Mon Sep 17 00:00:00 2001 From: Dustin Byrne Date: Tue, 3 Nov 2026 16:26:35 -0500 Subject: [PATCH 3/3] fix: preserve benchmark failure reports after timeouts --- CONTRIBUTING.md | 2 +- benchmarks/run.py | 40 +++++++++++++---------- benchmarks/test_benchmarks.py | 61 +++++++++++++++++++++++------------ 3 files changed, 64 insertions(+), 39 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 1efa9631..af5f8fc1 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -102,7 +102,7 @@ Measurement scope: - A coordinator-owned loopback receiver exercises the normal SDK HTTP, batching, byte-size splitting and background-worker paths. Receiver CPU, allocations and retained fixtures are outside the measured Ruby process. Refresh elapsed/allocations include the worker's loopback control request selecting the next generation. No production service or credentials are used. - Public local-only flag getters and reload APIs are used, with flag-called events disabled. The single-flag getter's deprecation warning occurs during warmup. Capture uses public `batch_size: 100000`, `max_queue_size: 100001`, `flush_interval_seconds: 3600` and `compress_request: false`. Native byte-size batching remains active. Warmup deliveries are verified before counters reset; sequences, payloads, users and final delivery are validated without a rescuing flush for delivery workloads. -Workers have a 120-second deadline. After two timeouts a coordinator finishes the current pair, saves partial diagnostics and fails. Each CI shard has a 12-minute job and a 10-minute measurement deadline, reserving time for upload. Collection has five minutes and rejects incomplete shard coverage. Run duration must be assessed from hosted CI, not concurrent local measurements. +Workers have a 120-second deadline. After two timeouts a coordinator finishes the current pair, saves partial diagnostics and fails. Results retain the complete selected workload inventory: unvisited workloads are marked `error` with a not-run reason and no samples, so the failed shard can still be collected and published. Each CI shard has a 12-minute job and a 10-minute measurement deadline, reserving time for upload. Collection has five minutes and rejects incomplete shard coverage. Run duration must be assessed from hosted CI, not concurrent local measurements. The shared reporter validates raw result schema v1 and produces readable tables plus `comparison.json`. Its comment shortlist requires at least a 5% median change and all six pairs agreeing in direction; it is an informational noise filter, not statistical significance. Missing and unsupported counters remain explicit. The read-only producer workflow is separate from the trusted `workflow_run` publisher; automatic PR comments become active only once that caller workflow is on the default branch. diff --git a/benchmarks/run.py b/benchmarks/run.py index c0192d18..c5354161 100644 --- a/benchmarks/run.py +++ b/benchmarks/run.py @@ -367,38 +367,44 @@ def main(args=None): for key in ("runtime", "runtime_version", "measurement_tools", "os", "arch"): if documents["base"]["execution"][key] != documents["head"]["execution"][key]: raise ValueError(f"Base/head {key} differ") - options.output.mkdir(parents=True, exist_ok=True) - - def save(): - for role, document in documents.items(): - temporary = options.output / f"{role}.json.tmp" - temporary.write_text(json.dumps(document, allow_nan=False, indent=2) + "\n") - temporary.replace(options.output / f"{role}.json") - - failures = timeouts = 0 - for spec in specs: - records = {} - for role in directories: - record = { + for document in documents.values(): + document["benchmarks"] = [ + { "id": spec["id"], "parameters": spec, "fixture_digest": digest( { "spec": spec, - "users": USERS, "definitions": ( definitions(spec["flag_count"]) if "flag_count" in spec else None ), + "users": USERS, } ), - "status": "ok", + "status": "error", + "reason": "Workload was not run: suite stopped before reaching it", "metrics": [], "runs": [], } - records[role] = record - documents[role]["benchmarks"].append(record) + for spec in specs + ] + options.output.mkdir(parents=True, exist_ok=True) + + def save(): + for role, document in documents.items(): + temporary = options.output / f"{role}.json.tmp" + temporary.write_text(json.dumps(document, allow_nan=False, indent=2) + "\n") + temporary.replace(options.output / f"{role}.json") + + failures = timeouts = 0 + for index, spec in enumerate(specs): + records = {} + for role in directories: + records[role] = documents[role]["benchmarks"][index] + records[role]["status"] = "ok" + del records[role]["reason"] for repetition in range(options.repetitions): for measurement_pass in spec["passes"]: for role in ("head", "base") if repetition % 2 else ("base", "head"): diff --git a/benchmarks/test_benchmarks.py b/benchmarks/test_benchmarks.py index 0ab29faa..7957c5ba 100644 --- a/benchmarks/test_benchmarks.py +++ b/benchmarks/test_benchmarks.py @@ -181,27 +181,46 @@ def fake_worker(installation, spec, measurement_pass): calls.append(installation.name) return {"status": "timeout", "pass": measurement_pass} - with tempfile.TemporaryDirectory() as t, patch.object( - run, "document_for", side_effect=lambda *a: copy.deepcopy(doc) - ), patch.object(run, "run_worker", side_effect=fake_worker): - code = run.main( - [ - "--base", - t, - "--head", - t, - "--build-directory", - t, - "--output", - t, - "--shard", - "flags-single", - ] - ) - self.assertEqual(code, 1) - self.assertEqual(calls, ["base", "head"]) - saved = json.loads((Path(t) / "head.json").read_text()) - self.assertEqual(saved["benchmarks"][0]["status"], "timeout") + for shard in (None, *run.load_shards()): + calls.clear() + with self.subTest( + shard=shard + ), tempfile.TemporaryDirectory() as t, patch.object( + run, "document_for", side_effect=lambda *a: copy.deepcopy(doc) + ), patch.object( + run, "run_worker", side_effect=fake_worker + ): + code = run.main( + [ + "--base", + t, + "--head", + t, + "--build-directory", + t, + "--output", + t, + *(["--shard", shard] if shard else []), + ] + ) + self.assertEqual(code, 1) + self.assertEqual(calls, ["base", "head"]) + expected = ( + run.load_shards()[shard] + if shard + else [spec["id"] for spec in fixtures.workloads()] + ) + for role in ("base", "head"): + saved = json.loads((Path(t) / f"{role}.json").read_text()) + records = saved["benchmarks"] + self.assertEqual([record["id"] for record in records], expected) + self.assertEqual(records[0]["status"], "timeout") + self.assertEqual(len(records[0]["runs"]), 1) + for record in records[1:]: + self.assertEqual(record["status"], "error") + self.assertIn("not run", record["reason"]) + self.assertEqual(record["metrics"], []) + self.assertEqual(record["runs"], []) def test_alternating_pairs(self): doc = {