From 4723bcab1218a12103c7e031366168dd8d8f1879 Mon Sep 17 00:00:00 2001 From: Connor Tsui Date: Fri, 19 Jun 2026 14:22:14 -0400 Subject: [PATCH 1/2] scripts: add v4 --postgres dual-write writer and measurement_id port Teach scripts/post-ingest.py a --postgres mode that connects to the v4 RDS Postgres (IAM auth, verify-full TLS) and upserts benchmark results directly via INSERT ... ON CONFLICT (measurement_id) DO UPDATE, in one transaction with a NaN/Inf guard and a best-effort site revalidate. The pre-existing v3 --server path is unchanged and stays stdlib-only; the v4 path lazily imports its third-party deps (psycopg[binary], boto3, xxhash) supplied at the call site via uv run --no-project --with, so importing the module or running --server never needs them. scripts/_measurement_id.py is a byte-for-byte port of the Rust measurement_id xxhash64 in vortex-data/benchmarks-website's server/src/db.rs. The writer computes the row key client-side so re-ingesting an existing (commit, dimension) upserts the existing row instead of inserting a duplicate. The port is verified against the golden vectors generated by that repo's measurement_id_golden_vectors test (65 vectors pass as of this commit); the vectors and the parity test live with the Rust source of truth rather than in this repo. Signed-off-by: Connor Tsui --- scripts/_measurement_id.py | 220 +++++++++ scripts/post-ingest.py | 969 +++++++++++++++++++++++++++++++++++-- 2 files changed, 1162 insertions(+), 27 deletions(-) create mode 100644 scripts/_measurement_id.py diff --git a/scripts/_measurement_id.py b/scripts/_measurement_id.py new file mode 100644 index 00000000000..e8834f4c097 --- /dev/null +++ b/scripts/_measurement_id.py @@ -0,0 +1,220 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright the Vortex contributors + +"""Python port of the server-internal `measurement_id` xxhash64 functions. + +This is a byte-for-byte port of the `vortex-data/benchmarks-website` repo's +`server/src/db.rs` (`measurement_id_*`, `hasher_for`, +`write_str`, `write_opt_str`, `write_i32`, `write_f64`, `finish`). The v4 ingest +writer computes the same `measurement_id` the v3 Rust server computes, so +re-ingesting an existing `(commit, dim-tuple)` upserts the existing row via +`ON CONFLICT (measurement_id) DO UPDATE` instead of inserting a duplicate. + +The equivalence is NOT assumed -- it is pinned by golden vectors generated from +the Rust source of truth (the `measurement_id_golden_vectors` test in the +`vortex-data/benchmarks-website` repo's `server/src/db.rs`). The Python port is +checked against those vectors; any drift in either implementation fails that +check. + +## The hash, precisely + +`measurement_id` is a canonical xxhash64 (seed 0) over a byte buffer built as: + +1. The per-table tag bytes, then a single `0x00` separator (`hasher_for`). +2. For each dimensional field, in the exact order the Rust function writes it: + - `write_str(s)` -> `len(utf8(s))` as a little-endian u64, then the utf8 + bytes. The length is the BYTE length, not the character count. + - `write_opt_str(o)` -> `0x00` for `None`; `0x01` then `write_str(s)` for + `Some(s)`. + - `write_i32(v)` -> 4 little-endian two's-complement bytes. + - `write_f64(v)` -> the IEEE-754 bit pattern as 8 little-endian bytes + (`v.to_bits()` written little-endian == `struct.pack(" i64` (Postgres/DuckDB `BIGINT` is + signed), matching Rust's `hasher.finish() as i64`. + +Canonical xxhash64 is streaming-equivalent: feeding the bytes incrementally (as +the Rust `Hasher` does via `write_*`) yields the same digest as hashing the +fully concatenated buffer once (as this module does). + +The little-endian integer encodings are load-bearing and assume a little-endian +host. The Rust side serializes integers with native-endian byte order +(twox-hash 2.x `write_u64` / `write_i32` use `to_ne_bytes`, and `write_f64` is +`write_u64(v.to_bits())`), so byte-for-byte compatibility holds only where +`to_ne_bytes == to_le_bytes`. Every target in play (x86_64 / aarch64 CI runners, +dev machines, the RDS Postgres host, the Vercel reader) is little-endian, and the +golden vectors are generated on a little-endian host and pin it there. On a +big-endian host both this module and the Rust `write_*` would have to switch to a +shared explicit endianness. +""" + +import struct + +import xxhash + +# Must match `XxHash64::with_seed(0)` in `db.rs::hasher_for`. +_SEED = 0 + +# Per-table tag literals. These MUST match the `hasher_for("")` argument in +# each `db.rs::measurement_id_*` function verbatim; the tag is the table name. +_TAG_QUERY_MEASUREMENTS = "query_measurements" +_TAG_COMPRESSION_TIMES = "compression_times" +_TAG_COMPRESSION_SIZES = "compression_sizes" +_TAG_RANDOM_ACCESS_TIMES = "random_access_times" +_TAG_VECTOR_SEARCH_RUNS = "vector_search_runs" + + +def _hasher_buf(tag: str) -> bytearray: + """Start a hash buffer seeded with a per-table tag plus a `0x00` separator. + + Mirrors `db.rs::hasher_for`: two fact tables that share the same dim values + still produce distinct `measurement_id`s because the tag differs. + """ + buf = bytearray() + buf += tag.encode("utf-8") + buf.append(0) + return buf + + +def _write_str(buf: bytearray, s: str) -> None: + """Append a length-prefixed string: utf8 BYTE length as LE u64, then bytes.""" + encoded = s.encode("utf-8") + buf += struct.pack(" None: + """Append an optional string: `0x00` for None, `0x01` + the string for Some.""" + if s is None: + buf.append(0) + else: + buf.append(1) + _write_str(buf, s) + + +def _write_i32(buf: bytearray, v: int) -> None: + """Append a 32-bit signed integer as 4 little-endian two's-complement bytes.""" + buf += struct.pack(" None: + """Append a 64-bit float as its 8 little-endian IEEE-754 bytes. + + `struct.pack(" int: + """Hash the buffer (xxhash64, seed 0) and bit-cast the u64 digest to i64.""" + digest = xxhash.xxh64(bytes(buf), seed=_SEED).intdigest() + # Bit-cast u64 -> i64 to match Rust's `hasher.finish() as i64`. + return digest - (1 << 64) if digest >= (1 << 63) else digest + + +def measurement_id_query( + *, + commit_sha: str, + dataset: str, + dataset_variant: str | None, + scale_factor: str | None, + query_idx: int, + storage: str, + engine: str, + format: str, +) -> int: + """`measurement_id` for a `query_measurements` row. Mirrors + `db.rs::measurement_id_query`.""" + buf = _hasher_buf(_TAG_QUERY_MEASUREMENTS) + _write_str(buf, commit_sha) + _write_str(buf, dataset) + _write_opt_str(buf, dataset_variant) + _write_opt_str(buf, scale_factor) + _write_i32(buf, query_idx) + _write_str(buf, storage) + _write_str(buf, engine) + _write_str(buf, format) + return _finish(buf) + + +def measurement_id_compression_time( + *, + commit_sha: str, + dataset: str, + dataset_variant: str | None, + format: str, + op: str, +) -> int: + """`measurement_id` for a `compression_times` row. Mirrors + `db.rs::measurement_id_compression_time`.""" + buf = _hasher_buf(_TAG_COMPRESSION_TIMES) + _write_str(buf, commit_sha) + _write_str(buf, dataset) + _write_opt_str(buf, dataset_variant) + _write_str(buf, format) + _write_str(buf, op) + return _finish(buf) + + +def measurement_id_compression_size( + *, + commit_sha: str, + dataset: str, + dataset_variant: str | None, + format: str, +) -> int: + """`measurement_id` for a `compression_sizes` row. Mirrors + `db.rs::measurement_id_compression_size`.""" + buf = _hasher_buf(_TAG_COMPRESSION_SIZES) + _write_str(buf, commit_sha) + _write_str(buf, dataset) + _write_opt_str(buf, dataset_variant) + _write_str(buf, format) + return _finish(buf) + + +def measurement_id_random_access( + *, + commit_sha: str, + dataset: str, + format: str, +) -> int: + """`measurement_id` for a `random_access_times` row. Mirrors + `db.rs::measurement_id_random_access`. Note: no `dataset_variant`.""" + buf = _hasher_buf(_TAG_RANDOM_ACCESS_TIMES) + _write_str(buf, commit_sha) + _write_str(buf, dataset) + _write_str(buf, format) + return _finish(buf) + + +def measurement_id_vector_search( + *, + commit_sha: str, + dataset: str, + layout: str, + flavor: str, + threshold: float, +) -> int: + """`measurement_id` for a `vector_search_runs` row. Mirrors + `db.rs::measurement_id_vector_search`. `iterations` is intentionally NOT part + of the dim tuple -- it is a side count.""" + buf = _hasher_buf(_TAG_VECTOR_SEARCH_RUNS) + _write_str(buf, commit_sha) + _write_str(buf, dataset) + _write_str(buf, layout) + _write_str(buf, flavor) + _write_f64(buf, threshold) + return _finish(buf) + + +# Dispatch table keyed by the fact-table name, used by the golden-vector test to +# map a vector's `table` field to the matching port function. Keeping it here (vs. +# in the test) means a new fact table is wired in one place alongside the port. +MEASUREMENT_ID_BY_TABLE = { + _TAG_QUERY_MEASUREMENTS: measurement_id_query, + _TAG_COMPRESSION_TIMES: measurement_id_compression_time, + _TAG_COMPRESSION_SIZES: measurement_id_compression_size, + _TAG_RANDOM_ACCESS_TIMES: measurement_id_random_access, + _TAG_VECTOR_SEARCH_RUNS: measurement_id_vector_search, +} diff --git a/scripts/post-ingest.py b/scripts/post-ingest.py index 2a72f975d95..e349048f04b 100755 --- a/scripts/post-ingest.py +++ b/scripts/post-ingest.py @@ -7,29 +7,44 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright the Vortex contributors -"""Wrap a `--gh-json-v3` JSONL file in an envelope and POST to /api/ingest. +"""Ingest a `--gh-json-v3` JSONL file into the benchmarks store. -Reads bare v3 records from a JSONL file produced by `vortex-bench --gh-json-v3`, -fills the `commit` envelope by shelling out to `git show`, and POSTs the -envelope to `/api/ingest` with a bearer token. +Reads bare v3 records from a JSONL file produced by `vortex-bench --gh-json-v3` +and fills the `commit` fields by shelling out to `git show`. Two mutually +exclusive ingest modes select the destination substrate: -Standard library only -- urllib, json, subprocess. The default envelope size -(60 MiB, just under the server's 64 MiB body limit) is sized so a single -JSONL run normally posts in one envelope -- preserving the "per-file -all-or-nothing" contract the server documents. If the JSONL is large enough -that splitting kicks in, the script emits a warning and proceeds with the +- `--server ` (v3): wraps the records in an envelope and POSTs to + `/api/ingest` with a bearer token. Standard library only -- urllib, + json, subprocess. +- `--postgres ` (v4): computes the server-internal `measurement_id` + locally (via `_measurement_id.py`) and upserts directly into the RDS + Postgres tables with `INSERT ... ON CONFLICT (measurement_id) DO UPDATE`, + over a verify-full TLS connection, authenticating with an RDS IAM auth token + when the DSN carries no password. Requires `psycopg`, `boto3`, and `xxhash` + from the project environment; these are imported lazily inside the + `--postgres` code path so the v3 `--server` path stays standard-library-only + under a bare `python3` (CI invokes `python3 scripts/post-ingest.py`, not + `uv run`, and the v3 path is in production until the Phase 5 cutover). The + PEP 723 metadata block below intentionally keeps `dependencies = []` for that + reason; run the `--postgres` mode from the repo's uv environment. + +The default envelope size (60 MiB, just under the server's 64 MiB body limit) +is sized so a single JSONL run normally posts in one envelope -- preserving the +"per-file all-or-nothing" contract the server documents. If the JSONL is large +enough that splitting kicks in, the script emits a warning and proceeds with the chunked semantics (per-chunk commit, mid-chunk failure leaves earlier chunks -ingested; subsequent retries re-upsert via the server's ON CONFLICT -idempotency on `measurement_id`). +ingested; subsequent retries re-upsert via the server's ON CONFLICT idempotency +on `measurement_id`). The `--postgres` mode applies a whole JSONL file in one +transaction (all-or-nothing), with no chunking. Wire-contract pointers (kept in sync as a coordinated change per -`benchmarks-website/AGENTS.md`): +the `vortex-data/benchmarks-website` repo's `AGENTS.md`): -- `benchmarks-website/server/src/records.rs` - envelope + per-record wire +- the `vortex-data/benchmarks-website` repo's `server/src/records.rs` - envelope + per-record wire shapes that the server deserializes. - `vortex-bench/src/v3.rs` - bare-record producer that writes the JSONL this script wraps. -- `benchmarks-website/server/src/schema.rs` - `SCHEMA_VERSION` source of +- the `vortex-data/benchmarks-website` repo's `server/src/schema.rs` - `SCHEMA_VERSION` source of truth that the `SCHEMA_VERSION` constant below MUST equal at every bump. """ @@ -37,17 +52,19 @@ import argparse import json +import math import os import subprocess import sys import urllib.error import urllib.request +from concurrent.futures import ThreadPoolExecutor from datetime import UTC, datetime from pathlib import Path -# MUST equal `benchmarks-website/server/src/schema.rs::SCHEMA_VERSION`. +# MUST equal the `vortex-data/benchmarks-website` repo's `server/src/schema.rs::SCHEMA_VERSION`. # Bumping this is a coordinated change across schema.rs, records.rs, v3.rs, -# and this script. See `benchmarks-website/AGENTS.md` ("Wire shapes are a +# and this script. See the `vortex-data/benchmarks-website` repo's `AGENTS.md` ("Wire shapes are a # coordinated change") for the full list of coupled sites. SCHEMA_VERSION = 1 # Default sized to fit comfortably under the server's 64 MiB ingest body @@ -59,17 +76,45 @@ def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( - description="POST a v3 JSONL records file to /api/ingest.", + description=( + "Ingest a v3 JSONL records file: --server POSTs a v3 envelope to " + "/api/ingest; --postgres upserts directly into the RDS Postgres tables." + ), ) parser.add_argument( "jsonl_path", type=Path, help="Path to the JSONL file written by vortex-bench --gh-json-v3.", ) - parser.add_argument( + mode = parser.add_mutually_exclusive_group(required=True) + mode.add_argument( "--server", - required=True, - help="Server base URL, e.g. http://localhost:8080.", + help=( + "v3 mode: server base URL, e.g. http://localhost:8080. Wraps the " + "records in an envelope and POSTs to /api/ingest with a " + "bearer token. Mutually exclusive with --postgres." + ), + ) + mode.add_argument( + "--postgres", + metavar="DSN", + help=( + "v4 mode: libpq DSN for the RDS Postgres ingest target, e.g. " + "'postgresql://bench_ingest@host:5432/benchmarks?sslmode=verify-full" + "&sslrootcert=/path/rds-ca.pem'. Computes measurement_id locally and " + "upserts via INSERT ... ON CONFLICT DO UPDATE. If the DSN carries no " + "password, an RDS IAM auth token is minted for the DSN's user. " + "Mutually exclusive with --server." + ), + ) + parser.add_argument( + "--region", + default=None, + help=( + "AWS region for RDS IAM token minting (--postgres mode). Precedence: " + "this explicit --region, then the boto3 session region, then the region " + "parsed from the RDS hostname." + ), ) parser.add_argument( "--commit-sha", @@ -78,8 +123,10 @@ def parse_args() -> argparse.Namespace: ) parser.add_argument( "--benchmark-id", - required=True, - help="Run identifier echoed back in run_meta.benchmark_id.", + default=None, + help="Run identifier echoed back in run_meta.benchmark_id. Required for " + "--server mode; unused by --postgres mode (the v4 tables have no " + "benchmark_id column).", ) parser.add_argument( "--token-env", @@ -113,9 +160,11 @@ def parse_args() -> argparse.Namespace: def read_records(path: Path) -> list[dict]: + # The JSONL comes from the project's own `vortex-bench` CI on the same run (trusted UTF-8 + # input); a malformed line still fails loud with `path:line` context for debuggability. records: list[dict] = [] - with path.open("r", encoding="utf-8") as fp: - for line_no, line in enumerate(fp, start=1): + with path.open(encoding="utf-8") as f: + for line_no, line in enumerate(f, start=1): line = line.strip() if not line: continue @@ -127,7 +176,10 @@ def read_records(path: Path) -> list[dict]: def git_show_field(sha: str, fmt: str, cwd: Path | None) -> str: - """Run `git show -s --format= ` and return its stdout (stripped).""" + """Run `git show -s --format= ` and return its stdout (stripped). + + The metadata comes from the project's own (UTF-8) git history. + """ result = subprocess.run( ["git", "show", "-s", f"--format={fmt}", sha], cwd=cwd, @@ -265,9 +317,865 @@ def post(server: str, body: bytes, token: str, timeout: float) -> tuple[int, byt return exc.code, exc.read() -def main() -> int: - args = parse_args() +# -------------------------------------------------------------------------- +# Postgres dual-write mode (--postgres) +# -------------------------------------------------------------------------- +# The v4 ingest path. `psycopg`, `boto3`, and `_measurement_id` (which pulls in +# `xxhash`) are imported lazily inside the functions below so the v3 `--server` +# path above stays standard-library-only under a bare `python3`. + +# Per-`kind` record field sets, mirroring the `#[serde(deny_unknown_fields)]` +# structs in the `vortex-data/benchmarks-website` repo's `server/src/records.rs`. `required` plus +# `optional` is the exact set of fields a record of that `kind` may carry +# (besides `kind` itself); an unknown field is rejected loudly, preserving the +# v3 server's deny_unknown_fields behavior so producer/schema drift surfaces +# instead of silently dropping data. `optional` fields default to NULL. +_RECORD_FIELDS: dict[str, tuple[frozenset[str], frozenset[str]]] = { + "query_measurement": ( + frozenset( + { + "commit_sha", + "dataset", + "query_idx", + "storage", + "engine", + "format", + "value_ns", + "all_runtimes_ns", + } + ), + frozenset( + { + "dataset_variant", + "scale_factor", + "peak_physical", + "peak_virtual", + "physical_delta", + "virtual_delta", + "env_triple", + } + ), + ), + "compression_time": ( + frozenset({"commit_sha", "dataset", "format", "op", "value_ns", "all_runtimes_ns"}), + frozenset({"dataset_variant", "env_triple"}), + ), + "compression_size": ( + frozenset({"commit_sha", "dataset", "format", "value_bytes"}), + frozenset({"dataset_variant"}), + ), + "random_access_time": ( + frozenset({"commit_sha", "dataset", "format", "value_ns", "all_runtimes_ns"}), + frozenset({"env_triple"}), + ), + "vector_search_run": ( + frozenset( + { + "commit_sha", + "dataset", + "layout", + "flavor", + "threshold", + "value_ns", + "all_runtimes_ns", + "matches", + "rows_scanned", + "bytes_scanned", + "iterations", + } + ), + frozenset({"env_triple"}), + ), +} + +_MEASUREMENT_ID_MODULE = None + + +def _measurement_id_module(): + """Lazily load `scripts/_measurement_id.py` by path (cached). + + Loaded by file path rather than `import _measurement_id` so it resolves + regardless of cwd / `sys.path` (the test harness loads sibling scripts the + same way). Importing it pulls in `xxhash`, so it only happens here on the + `--postgres` path. + """ + global _MEASUREMENT_ID_MODULE + if _MEASUREMENT_ID_MODULE is None: + import importlib.util + + path = Path(__file__).resolve().parent / "_measurement_id.py" + spec = importlib.util.spec_from_file_location("_measurement_id", path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + _MEASUREMENT_ID_MODULE = module + return _MEASUREMENT_ID_MODULE + + +def _validate_record_fields(record: object, index: int) -> str: + """Validate a record's `kind` and field set; return the `kind`. + + Mirrors the server's deny_unknown_fields + required-field deserialization: + a non-object record, an unknown `kind`, an unknown field, or a missing + required field is a loud error keyed by the record's index. + """ + if not isinstance(record, dict): + raise SystemExit(f"record {index}: expected a JSON object, got {type(record).__name__}") + kind = record.get("kind") + # `isinstance(kind, str)` first: a non-scalar kind (e.g. a list) is unhashable + # and would raise TypeError at the `in` membership check rather than this + # controlled record-indexed error. + if not isinstance(kind, str) or kind not in _RECORD_FIELDS: + raise SystemExit(f"record {index}: unknown kind {kind!r}; expected one of {sorted(_RECORD_FIELDS)}") + required, optional = _RECORD_FIELDS[kind] + present = set(record) - {"kind"} + unknown = present - required - optional + if unknown: + raise SystemExit(f"record {index} ({kind}): unknown field(s) {sorted(unknown)} (deny_unknown_fields)") + missing = required - present + if missing: + raise SystemExit(f"record {index} ({kind}): missing required field(s) {sorted(missing)}") + return kind + + +def _require_finite(value: object, field: str, kind: str, index: int) -> None: + """Raise loudly unless `value` is a finite number. + + Guards every f64 dim that feeds a `measurement_id` hash (currently only + `vector_search_run.threshold`). A NaN/Inf would hash differently across the + Rust `to_bits()` and Python `struct.pack(' None: + """Raise loudly unless `value` is a plain (non-bool) integer within i32/i64 range. + + The integer columns bind straight to INTEGER/BIGINT; psycopg adapts a Python + float to float8 and Postgres assignment-casts (rounds) it, so a JSON float + would silently persist a rounded value where the v3 server's serde `i32`/`i64` + rejects it. An out-of-range integer would otherwise fail late as an uncaught + `struct.error` (i32 hash dims via `_write_i32`) or a raw Postgres 22003 + overflow; validate the type AND width here so a malformed scalar fails loud + (record-indexed) instead, matching the v3 serde boundary. + """ + lo, hi = (_INT32_MIN, _INT32_MAX) if bits == 32 else (_INT64_MIN, _INT64_MAX) + if isinstance(value, bool) or not isinstance(value, int): + raise SystemExit(f"record {index} ({kind}): {field} must be an integer, got {value!r}") + if not (lo <= value <= hi): + raise SystemExit(f"record {index} ({kind}): {field}={value!r} is out of int{bits} range") + + +def _require_int_list(value: object, field: str, kind: str, index: int) -> None: + """Raise loudly unless `value` is a JSON array of plain (non-bool) i64 integers. + + Guards the `all_runtimes_ns` -> `bigint[]` bind: the explicit `::bigint[]` + cast is permissive in ways the v3 server's `Vec` serde is not. psycopg + sends the string `"{}"` as text and the cast parses it into an empty array, a + `[1, null]` list adapts to `{1,NULL}`, and an out-of-i64 element hits a raw + Postgres 22003 -- each diverges from or fails differently than the v3 path. + Validate the element type AND i64 range so a malformed value fails loud. + """ + if not isinstance(value, list) or any(isinstance(x, bool) or not isinstance(x, int) for x in value): + raise SystemExit(f"record {index} ({kind}): {field} must be a JSON array of integers, got {value!r}") + if any(not (_INT64_MIN <= x <= _INT64_MAX) for x in value): + raise SystemExit(f"record {index} ({kind}): {field} has an element out of int64 range") + + +def _require_str(value: object, field: str, kind: str, index: int) -> None: + """Raise loudly unless `value` is a string. Mirrors the v3 serde `String` fields.""" + if not isinstance(value, str): + raise SystemExit(f"record {index} ({kind}): {field} must be a string, got {value!r}") + + +def _require_opt_str(value: object, field: str, kind: str, index: int) -> None: + """Raise loudly unless `value` is a string or null. Mirrors `Option`.""" + if value is not None and not isinstance(value, str): + raise SystemExit(f"record {index} ({kind}): {field} must be a string or null, got {value!r}") + + +def _memory_quartet_consistent(r: dict) -> bool: + """The four `query_measurements` memory columns are all set or all absent. + + Mirrors `ingest.rs::memory_quartet_consistent`: a partial quartet is a + validation error, not a half-populated row. + """ + present = [ + r.get("peak_physical") is not None, + r.get("peak_virtual") is not None, + r.get("physical_delta") is not None, + r.get("virtual_delta") is not None, + ] + return not any(present) or all(present) + + +# Per-kind field -> type token, mirroring the v3 server's typed serde boundary in +# `records.rs`. The direct-to-Postgres writer can no longer lean on server-side +# deserialization, so every field is type/range-validated here. Tokens: +# "str" required String "opt_str" Option +# "i32" required i32 "i64" required i64 +# "opt_i64" Option "i64_list" Vec (all_runtimes_ns) +# "f64" finite f64 (threshold). +# Required tokens (every non-opt_*) line up with `_RECORD_FIELDS` required sets, so +# `record[field]` is always present by the time `_validate_record_values` runs. +_FIELD_TYPES: dict[str, tuple[tuple[str, str], ...]] = { + "query_measurement": ( + ("commit_sha", "str"), + ("dataset", "str"), + ("dataset_variant", "opt_str"), + ("scale_factor", "opt_str"), + ("query_idx", "i32"), + ("storage", "str"), + ("engine", "str"), + ("format", "str"), + ("value_ns", "i64"), + ("all_runtimes_ns", "i64_list"), + ("peak_physical", "opt_i64"), + ("peak_virtual", "opt_i64"), + ("physical_delta", "opt_i64"), + ("virtual_delta", "opt_i64"), + ("env_triple", "opt_str"), + ), + "compression_time": ( + ("commit_sha", "str"), + ("dataset", "str"), + ("dataset_variant", "opt_str"), + ("format", "str"), + ("op", "str"), + ("value_ns", "i64"), + ("all_runtimes_ns", "i64_list"), + ("env_triple", "opt_str"), + ), + "compression_size": ( + ("commit_sha", "str"), + ("dataset", "str"), + ("dataset_variant", "opt_str"), + ("format", "str"), + ("value_bytes", "i64"), + ), + "random_access_time": ( + ("commit_sha", "str"), + ("dataset", "str"), + ("format", "str"), + ("value_ns", "i64"), + ("all_runtimes_ns", "i64_list"), + ("env_triple", "opt_str"), + ), + "vector_search_run": ( + ("commit_sha", "str"), + ("dataset", "str"), + ("layout", "str"), + ("flavor", "str"), + ("threshold", "f64"), + ("value_ns", "i64"), + ("all_runtimes_ns", "i64_list"), + ("matches", "i64"), + ("rows_scanned", "i64"), + ("bytes_scanned", "i64"), + ("iterations", "i32"), + ("env_triple", "opt_str"), + ), +} + + +def _validate_record_values(record: dict, kind: str, index: int) -> None: + """Validate every field's type/range against the v3 server's serde boundary. + + Runs in `ingest_postgres`'s loop (where the record index is known) so every + failure is reported as `record {index} ({kind}): ...`, matching the v3 + server's indexed per-record errors. Drives field type/range checks off + `_FIELD_TYPES`, then applies the semantic checks the type alone does not cover + (the storage enum + memory quartet for query_measurements). + """ + for field, typ in _FIELD_TYPES[kind]: + if typ == "str": + _require_str(record[field], field, kind, index) + elif typ == "opt_str": + _require_opt_str(record.get(field), field, kind, index) + elif typ == "i32": + _require_int(record[field], field, kind, index, bits=32) + elif typ == "i64": + _require_int(record[field], field, kind, index, bits=64) + elif typ == "opt_i64": + if record.get(field) is not None: + _require_int(record[field], field, kind, index, bits=64) + elif typ == "i64_list": + _require_int_list(record[field], field, kind, index) + elif typ == "f64": + _require_finite(record[field], field, kind, index) + + if kind == "query_measurement": + if record["storage"] not in ("nvme", "s3"): + raise SystemExit( + f"record {index} (query_measurement): storage must be 'nvme' or 's3', got {record['storage']!r}" + ) + if not _memory_quartet_consistent(record): + raise SystemExit( + f"record {index} (query_measurement): memory fields must be populated together (all four or none)" + ) + + +def _upsert_returning_was_update(conn, sql: str, params: tuple) -> bool: + """Run an `INSERT ... ON CONFLICT DO UPDATE ... RETURNING (xmax = 0)` upsert + and return whether the row was an update (vs a fresh insert). + + Classifies inserted-vs-updated atomically from the upsert itself: a fresh + INSERT leaves the new tuple's system column `xmax = 0`, while `ON CONFLICT + DO UPDATE` stamps `xmax` with the current transaction id. A preflight SELECT + (the prior approach) would race the upsert under concurrent re-ingest of the + same `measurement_id`, miscounting a concurrent loser as inserted; deriving + the flag from the single atomic statement removes that window. A duplicate + `measurement_id` within ONE transaction is also handled correctly: the second + upsert locks the tuple the first created (stamping its xmax), so it is + classified as an update, matching the v3 server. `sql` must end with + `RETURNING (xmax = 0) AS inserted`. + """ + row = conn.execute(sql, params).fetchone() + return not row[0] + + +def _insert_query_measurement(conn, mid_mod, r: dict) -> bool: + """Upsert a `query_measurements` row. Mirrors `ingest.rs::insert_query_measurement`. + + Record values are validated by `_validate_record_values` before dispatch. + """ + mid = mid_mod.measurement_id_query( + commit_sha=r["commit_sha"], + dataset=r["dataset"], + dataset_variant=r.get("dataset_variant"), + scale_factor=r.get("scale_factor"), + query_idx=r["query_idx"], + storage=r["storage"], + engine=r["engine"], + format=r["format"], + ) + return _upsert_returning_was_update( + conn, + """ + INSERT INTO query_measurements ( + measurement_id, commit_sha, dataset, dataset_variant, scale_factor, + query_idx, storage, engine, format, + value_ns, all_runtimes_ns, + peak_physical, peak_virtual, physical_delta, virtual_delta, + env_triple, commit_timestamp + ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::bigint[], %s, %s, %s, %s, %s, + (SELECT timestamp FROM commits WHERE commit_sha = %s)) + ON CONFLICT (measurement_id) DO UPDATE SET + commit_sha = excluded.commit_sha, + value_ns = excluded.value_ns, + all_runtimes_ns = excluded.all_runtimes_ns, + peak_physical = excluded.peak_physical, + peak_virtual = excluded.peak_virtual, + physical_delta = excluded.physical_delta, + virtual_delta = excluded.virtual_delta, + env_triple = excluded.env_triple, + commit_timestamp = excluded.commit_timestamp + RETURNING (xmax = 0) AS inserted + """, + ( + mid, + r["commit_sha"], + r["dataset"], + r.get("dataset_variant"), + r.get("scale_factor"), + r["query_idx"], + r["storage"], + r["engine"], + r["format"], + r["value_ns"], + r["all_runtimes_ns"], + r.get("peak_physical"), + r.get("peak_virtual"), + r.get("physical_delta"), + r.get("virtual_delta"), + r.get("env_triple"), + # The denormalized `commit_timestamp` (migration 006) is resolved from the + # `commits` row this same transaction upserted first, so the read path's + # latest-per-series summary never sees a NULL from this writer. + r["commit_sha"], + ), + ) + + +def _insert_compression_time(conn, mid_mod, r: dict) -> bool: + """Upsert a `compression_times` row. Mirrors `ingest.rs::insert_compression_time`.""" + mid = mid_mod.measurement_id_compression_time( + commit_sha=r["commit_sha"], + dataset=r["dataset"], + dataset_variant=r.get("dataset_variant"), + format=r["format"], + op=r["op"], + ) + return _upsert_returning_was_update( + conn, + """ + INSERT INTO compression_times ( + measurement_id, commit_sha, dataset, dataset_variant, + format, op, value_ns, all_runtimes_ns, env_triple + ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s::bigint[], %s) + ON CONFLICT (measurement_id) DO UPDATE SET + commit_sha = excluded.commit_sha, + value_ns = excluded.value_ns, + all_runtimes_ns = excluded.all_runtimes_ns, + env_triple = excluded.env_triple + RETURNING (xmax = 0) AS inserted + """, + ( + mid, + r["commit_sha"], + r["dataset"], + r.get("dataset_variant"), + r["format"], + r["op"], + r["value_ns"], + r["all_runtimes_ns"], + r.get("env_triple"), + ), + ) + + +def _insert_compression_size(conn, mid_mod, r: dict) -> bool: + """Upsert a `compression_sizes` row. Mirrors `ingest.rs::insert_compression_size`.""" + mid = mid_mod.measurement_id_compression_size( + commit_sha=r["commit_sha"], + dataset=r["dataset"], + dataset_variant=r.get("dataset_variant"), + format=r["format"], + ) + return _upsert_returning_was_update( + conn, + """ + INSERT INTO compression_sizes ( + measurement_id, commit_sha, dataset, dataset_variant, + format, value_bytes + ) VALUES (%s, %s, %s, %s, %s, %s) + ON CONFLICT (measurement_id) DO UPDATE SET + commit_sha = excluded.commit_sha, + value_bytes = excluded.value_bytes + RETURNING (xmax = 0) AS inserted + """, + ( + mid, + r["commit_sha"], + r["dataset"], + r.get("dataset_variant"), + r["format"], + r["value_bytes"], + ), + ) + + +def _insert_random_access(conn, mid_mod, r: dict) -> bool: + """Upsert a `random_access_times` row. Mirrors `ingest.rs::insert_random_access`.""" + mid = mid_mod.measurement_id_random_access( + commit_sha=r["commit_sha"], + dataset=r["dataset"], + format=r["format"], + ) + return _upsert_returning_was_update( + conn, + """ + INSERT INTO random_access_times ( + measurement_id, commit_sha, dataset, format, + value_ns, all_runtimes_ns, env_triple + ) VALUES (%s, %s, %s, %s, %s, %s::bigint[], %s) + ON CONFLICT (measurement_id) DO UPDATE SET + commit_sha = excluded.commit_sha, + value_ns = excluded.value_ns, + all_runtimes_ns = excluded.all_runtimes_ns, + env_triple = excluded.env_triple + RETURNING (xmax = 0) AS inserted + """, + ( + mid, + r["commit_sha"], + r["dataset"], + r["format"], + r["value_ns"], + r["all_runtimes_ns"], + r.get("env_triple"), + ), + ) + +def _insert_vector_search(conn, mid_mod, r: dict) -> bool: + """Upsert a `vector_search_runs` row. Mirrors `ingest.rs::insert_vector_search`. + + `threshold` is validated finite by `_validate_record_values` before dispatch. + """ + threshold = float(r["threshold"]) + mid = mid_mod.measurement_id_vector_search( + commit_sha=r["commit_sha"], + dataset=r["dataset"], + layout=r["layout"], + flavor=r["flavor"], + threshold=threshold, + ) + return _upsert_returning_was_update( + conn, + """ + INSERT INTO vector_search_runs ( + measurement_id, commit_sha, dataset, layout, flavor, threshold, + value_ns, all_runtimes_ns, matches, rows_scanned, bytes_scanned, + iterations, env_triple + ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s::bigint[], %s, %s, %s, %s, %s) + ON CONFLICT (measurement_id) DO UPDATE SET + commit_sha = excluded.commit_sha, + value_ns = excluded.value_ns, + all_runtimes_ns = excluded.all_runtimes_ns, + matches = excluded.matches, + rows_scanned = excluded.rows_scanned, + bytes_scanned = excluded.bytes_scanned, + iterations = excluded.iterations, + env_triple = excluded.env_triple + RETURNING (xmax = 0) AS inserted + """, + ( + mid, + r["commit_sha"], + r["dataset"], + r["layout"], + r["flavor"], + threshold, + r["value_ns"], + r["all_runtimes_ns"], + r["matches"], + r["rows_scanned"], + r["bytes_scanned"], + r["iterations"], + r.get("env_triple"), + ), + ) + + +# Dispatch from a record's `kind` to its per-table upsert. Keyed identically to +# `_RECORD_FIELDS`; the two maps are wired together when adding a fact table. +_APPLY_RECORD = { + "query_measurement": _insert_query_measurement, + "compression_time": _insert_compression_time, + "compression_size": _insert_compression_size, + "random_access_time": _insert_random_access, + "vector_search_run": _insert_vector_search, +} + + +def _upsert_commit(conn, commit: dict) -> None: + """Upsert the `commits` dim row. Mirrors `ingest.rs::upsert_commit`.""" + conn.execute( + """ + INSERT INTO commits ( + commit_sha, timestamp, message, author_name, author_email, + committer_name, committer_email, tree_sha, url + ) VALUES (%s, %s::timestamptz, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT (commit_sha) DO UPDATE SET + timestamp = excluded.timestamp, + message = excluded.message, + author_name = excluded.author_name, + author_email = excluded.author_email, + committer_name = excluded.committer_name, + committer_email = excluded.committer_email, + tree_sha = excluded.tree_sha, + url = excluded.url + """, + ( + commit["sha"], + commit["timestamp"], + commit["message"], + commit["author_name"], + commit["author_email"], + commit["committer_name"], + commit["committer_email"], + commit["tree_sha"], + commit["url"], + ), + ) + + +# Mirrors the v3 server's `WRITE_CONFLICT_ATTEMPTS` +# (the `vortex-data/benchmarks-website` repo's `server/src/ingest.rs`). The DuckDB ingest wrapped +# `apply_envelope_once` in `retry_write_conflicts`; the DuckDB -> Postgres substrate change +# must preserve that behavior, since the dual-write CI runs ~14 concurrent writers. +_WRITE_CONFLICT_ATTEMPTS = 128 + + +def _retry_write_conflicts(op): + """Retry `op` on a Postgres write conflict, mirroring the v3 server's `retry_write_conflicts`. + + The v3 DuckDB ingest retried on write conflicts up to `WRITE_CONFLICT_ATTEMPTS` times; the + Postgres writer must preserve that, because the dual-write CI runs many concurrent writers + whose row-level `ON CONFLICT DO UPDATE` upserts touching the same `commits` / dim rows in + conflicting orders can deadlock. The retryable analogs on Postgres are deadlock + (`SQLSTATE 40P01`) and serialization failure (`40001`); both abort one transaction cleanly, + so re-running the whole transaction is safe. A non-retryable error (e.g. a validation + `SystemExit`) propagates immediately. Returns `op`'s value on the first success. + """ + from psycopg import errors as pg_errors + + for attempt in range(1, _WRITE_CONFLICT_ATTEMPTS + 1): + try: + return op() + except (pg_errors.DeadlockDetected, pg_errors.SerializationFailure): + # The failing `op`'s `with conn.transaction()` block already rolled back, so the + # connection is idle and the whole transaction can be retried. Re-raise on the + # final attempt (mirrors v3 returning the error after the last try; the bare + # retry loop mirrors v3's `std::thread::yield_now()` between attempts). + if attempt >= _WRITE_CONFLICT_ATTEMPTS: + raise + raise AssertionError("unreachable: _retry_write_conflicts exited without return or raise") + + +def ingest_postgres(conn, commit: dict, records: list[dict]) -> tuple[int, int]: + """Upsert a commit and its records into Postgres, retrying on write conflicts. + + Wraps `_ingest_postgres_once` in `_retry_write_conflicts` (mirroring the v3 server's + `apply_envelope = retry_write_conflicts(apply_envelope_once)`). Returns `(inserted, updated)` + aggregated across all fact tables. + """ + mid_mod = _measurement_id_module() + return _retry_write_conflicts(lambda: _ingest_postgres_once(conn, commit, records, mid_mod)) + + +def _ingest_postgres_once(conn, commit: dict, records: list[dict], mid_mod) -> tuple[int, int]: + """Upsert a commit and its records into Postgres in one transaction (a single attempt). + + Mirrors the v3 server's `apply_envelope_once`: upsert `commits` first, then each fact + record, classifying each as inserted or updated. Any validation failure rolls the whole + transaction back (all-or-nothing). Returns `(inserted, updated)` aggregated across all + fact tables. + """ + inserted = 0 + updated = 0 + with conn.transaction(): + _upsert_commit(conn, commit) + for idx, record in enumerate(records): + kind = _validate_record_fields(record, idx) + if record["commit_sha"] != commit["sha"]: + raise SystemExit( + f"record {idx} ({kind}): commit_sha {record['commit_sha']!r} does not " + f"match envelope commit.sha {commit['sha']!r}" + ) + _validate_record_values(record, kind, idx) + if _APPLY_RECORD[kind](conn, mid_mod, record): + updated += 1 + else: + inserted += 1 + return inserted, updated + + +def _region_from_host(host: str) -> str | None: + """Parse the AWS region out of an RDS endpoint hostname. + + RDS endpoints look like `...rds.amazonaws.com` (instance) + or `.proxy-..rds.amazonaws.com` (proxy); the region is the + label immediately before `rds.amazonaws.com`. Returns None for any other + shape. + """ + parts = host.split(".") + if len(parts) >= 4 and parts[-3:] == ["rds", "amazonaws", "com"]: + return parts[-4] + return None + + +def _rds_iam_token(*, host: str, port: int, user: str, region: str | None) -> str: + """Mint a short-lived RDS IAM auth token to use as the connection password.""" + import boto3 + + session = boto3.session.Session() + resolved = region or session.region_name or _region_from_host(host) + if not resolved: + raise SystemExit( + "could not determine the AWS region for the RDS IAM token; pass --region or set AWS_DEFAULT_REGION." + ) + client = session.client("rds", region_name=resolved) + return client.generate_db_auth_token(DBHostname=host, Port=port, DBUsername=user, Region=resolved) + + +# The least-privilege login role the v4 CI ingest path must authenticate as +# (created by `migrations/004_ingest_role.sql`; SELECT,INSERT,UPDATE only). +# Phase-2 BAN: do not authenticate the ingest write path as `migrator` / +# `GitHubBenchmarkSchemaRole`. +_INGEST_ROLE = "bench_ingest" + + +def connect_postgres(dsn: str, region: str | None): + """Open a psycopg connection to the RDS Postgres ingest target. + + Enforces the ingest contract: verify-full TLS, and authentication only as the + least-privilege `bench_ingest` role -- always, regardless of auth method (IAM + token or password) or host. This is the production RDS ingest connector; the + test suite exercises ingest via `ingest_postgres` directly, so there is no + local-affordance exception to carve out. A host-based "is this local?" + heuristic would be unreliable anyway because libpq can resolve the host from a + DSN `hostaddr=` or the `$PGHOST` environment variable, either of which could + bypass a DSN-host check. Parses `dsn`; if it carries no password, mints an RDS + IAM auth token for the DSN's user and host (the DSN is expected to also supply + `sslrootcert` for verify-full to validate against). + """ + import psycopg + from psycopg import conninfo + + params = conninfo.conninfo_to_dict(dsn) + + # verify-full is the ingest TLS contract. Default an absent value, but + # refuse a DSN that explicitly downgrades it rather than silently weakening + # the internet-reachable ingest connection. + sslmode = params.get("sslmode") + if sslmode is None: + params["sslmode"] = "verify-full" + elif sslmode != "verify-full": + raise SystemExit( + f"--postgres requires sslmode=verify-full for the RDS ingest path; DSN " + f"specified sslmode={sslmode!r}. Omit it (defaults to verify-full) or set " + f"it to verify-full." + ) + + user = params.get("user") + # Least-privilege: always the bench_ingest role. No host heuristic (see the + # docstring): a misconfigured DSN -- by host=, hostaddr=, or $PGHOST -- must + # never ingest as migrator/postgres. + if user != _INGEST_ROLE: + raise SystemExit(f"--postgres must connect as the least-privilege {_INGEST_ROLE!r} role; DSN user is {user!r}.") + + if not params.get("password"): + # IAM-token path (production CI ingest): mint a token for the DSN's user. + host = params.get("host") + if not host: + raise SystemExit("--postgres DSN must specify host for IAM token minting") + try: + port = int(params.get("port", 5432)) + except (TypeError, ValueError) as exc: + raise SystemExit(f"--postgres DSN has a non-numeric port: {params.get('port')!r}") from exc + params["password"] = _rds_iam_token(host=host, port=port, user=user, region=region) + + # Force `search_path=public` so the writer's unqualified table names always resolve to the + # migration-owned `public.*` tables, regardless of any `search_path` baked into the DSN's + # `options=-c search_path=...` or the role's default. libpq applies repeated `-c` settings + # left-to-right (last wins), so appending ours last makes it authoritative even if the DSN + # already set one. + existing_options = params.get("options") or "" + params["options"] = f"{existing_options} -c search_path=public".strip() + + conn = psycopg.connect(**params) + # Verify the RESOLVED transport actually used TLS, not merely that the DSN requested + # verify-full. The `sslmode` check above rejects an explicit downgrade, but it cannot stop a + # hostless / Unix-socket DSN (host omitted, `host=/...`, or libpq resolving via `$PGHOST`) + # from connecting over a local socket, where libpq silently ignores `sslmode`. Checking + # `ssl_in_use` post-connect validates the connection that actually happened rather than + # trusting the DSN string -- the same "verify the resolved state" reasoning the docstring + # uses to reject a DSN-host heuristic -- and so also closes the `$PGHOST` / `hostaddr` bypass. + # NOTE: `ssl_in_use` lives on the low-level libpq wrapper `conn.pgconn` (a `pq.PGconn`); it is + # NOT on the high-level `conn.info` (`ConnectionInfo`), which exposes only host/dbname/user/etc. + if not conn.pgconn.ssl_in_use: + conn.close() + raise SystemExit( + "--postgres requires a verify-full TLS connection, but the established connection is " + "not using TLS (a hostless or Unix-socket DSN bypasses sslmode); connect to the RDS " + "instance over TCP with sslmode=verify-full." + ) + return conn + + +def _http(method: str, url: str, token: str | None, timeout: float) -> bytes: + """Issue one HTTP request and return the body. Raises on any non-2xx or + transport error; callers in `refresh_site_cache` swallow those.""" + headers = {"accept": "application/json"} + if token is not None: + headers["authorization"] = f"Bearer {token}" + req = urllib.request.Request(url, method=method, headers=headers) + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.read() + + +def _warm_default_windows(base: str, timeout: float) -> None: + """Best-effort warm pass: prime the freshly invalidated Data Cache for the + landing page and every group's default last-100 bundle, so the first human + request after an ingest is already hot. Each request is independent; one + failure does not abort the others.""" + + def warm(url: str) -> None: + try: + _http("GET", url, None, timeout) + except Exception as exc: # noqa: BLE001 -- warm is best-effort. + print(f"warning: warm {url} failed: {exc}", file=sys.stderr) + + warm(f"{base}/") + try: + groups_body = _http("GET", f"{base}/api/groups", None, timeout) + slugs = [g["slug"] for g in json.loads(groups_body).get("groups", []) if "slug" in g] + except Exception as exc: # noqa: BLE001 -- group discovery is best-effort. + print(f"warning: warm group discovery failed: {exc}", file=sys.stderr) + return + # A whole-bundle recompute is a few seconds cold, so warm with bounded + # concurrency rather than one slow serial pass. + with ThreadPoolExecutor(max_workers=4) as pool: + pool.map(lambda s: warm(f"{base}/api/group/{s}?n=100"), slugs) + + +def refresh_site_cache(base_url: str, token: str, timeout: float) -> None: + """Revalidate the site's Data Cache tag, then warm the default windows. + + BEST-EFFORT: every failure is logged to stderr and swallowed so a cache + refresh can never change the ingest exit code. The warm pass is skipped + when revalidation fails: warming after a failed flush would repopulate the + Data Cache with stale data, which is the opposite of the intent. + """ + base = base_url.rstrip("/") + try: + _http("POST", f"{base}/api/revalidate", token, timeout) + except Exception as exc: # noqa: BLE001 -- refresh must never raise into ingest. + print(f"warning: cache revalidate failed: {exc}", file=sys.stderr) + return # Skip the warm pass: no point warming a cache that was not flushed. + _warm_default_windows(base, timeout) + + +def _main_postgres(args: argparse.Namespace) -> int: + records = read_records(args.jsonl_path) + # `build_commit` runs `git show `, so the SHA must be in the runner's local git + # history. The v4 ingest step inherits the v3 `--server` step's checkout assumption (the default + # checkout provides the head SHA); a shallow checkout missing the SHA fails loud here, and the + # v4 step is best-effort (continue-on-error), so it never fails the job. + commit = build_commit(args.commit_sha, args.repo_url, args.git_dir) + conn = connect_postgres(args.postgres, args.region) + try: + inserted, updated = ingest_postgres(conn, commit, records) + finally: + conn.close() + print( + json.dumps( + {"records": len(records), "inserted": inserted, "updated": updated}, + separators=(",", ":"), + ) + ) + # Best-effort site-cache refresh after a successful write. No-op unless both + # env vars are set (so the script stays inert until the ops wiring lands), + # and it can never fail the ingest. The ops prerequisite (setting the two env + # vars in Vercel and as GitHub secrets/vars) is documented in the "Ops + # prerequisite" section of .big-plans/ct__bench-v4-uiux-r3-design.md. + base_url = os.environ.get("BENCH_SITE_BASE_URL") + revalidate_token = os.environ.get("BENCH_REVALIDATE_TOKEN") + if base_url and revalidate_token: + refresh_site_cache(base_url, revalidate_token, args.timeout) + return 0 + + +def _main_server(args: argparse.Namespace) -> int: + if args.benchmark_id is None: + print("error: --benchmark-id is required in --server mode", file=sys.stderr) + return 2 token = os.environ.get(args.token_env) if not token: print( @@ -336,5 +1244,12 @@ def main() -> int: return 0 +def main() -> int: + args = parse_args() + if args.postgres is not None: + return _main_postgres(args) + return _main_server(args) + + if __name__ == "__main__": raise SystemExit(main()) From aa5a15a39f222b69cd87b52d476faa091383b4c8 Mon Sep 17 00:00:00 2001 From: Connor Tsui Date: Fri, 19 Jun 2026 14:22:16 -0400 Subject: [PATCH 2/2] ci: add dormant best-effort v4 Postgres dual-write to the emitter workflows Add the v4 ingest step block to bench.yml, sql-benchmarks.yml, and v3-commit-metadata.yml (the last also gains id-token: write). Each block assumes the ingest role via OIDC and runs post-ingest.py --postgres after the existing v3 --server step. Every v4 step is gated on vars.GH_BENCH_INGEST_ROLE_ARN != '' and carries continue-on-error: true, so with the variable unset the code is dormant and a v4 failure can never fail a workflow or disturb the v2/v3 paths. Setting the variable is the later live cutover. Signed-off-by: Connor Tsui --- .github/workflows/bench.yml | 40 ++++++++++++++++++++++++ .github/workflows/sql-benchmarks.yml | 39 +++++++++++++++++++++++ .github/workflows/v3-commit-metadata.yml | 39 +++++++++++++++++++++++ 3 files changed, 118 insertions(+) diff --git a/.github/workflows/bench.yml b/.github/workflows/bench.yml index 146f0946dc0..cc26c6975ad 100644 --- a/.github/workflows/bench.yml +++ b/.github/workflows/bench.yml @@ -128,6 +128,46 @@ jobs: --benchmark-id "${{ matrix.benchmark.id }}" \ --repo-url "${{ github.server_url }}/${{ github.repository }}" + # v4 (Postgres) dual-write -- BEST-EFFORT during the migration soak. The + # proven v3 step above is hard-required; a v4 failure must NOT fail the job + # (v4 is promoted to required at cutover, PR-5.1). Gated on the ingest-role + # ARN var (the assume-role input that MUST exist for OIDC to succeed) so it + # no-ops until v4 infra is wired, and every step is + # `continue-on-error` so an OIDC / uv / connect hiccup never breaks the v3 + # pipeline. post-ingest.py mints the RDS IAM token internally (boto3) from + # the assumed GitHubBenchmarkIngestRole; sslmode=verify-full validates the cert. + - name: Configure AWS credentials for v4 ingest (OIDC) + if: vars.GH_BENCH_INGEST_ROLE_ARN != '' + continue-on-error: true + uses: aws-actions/configure-aws-credentials@e7f100cf4c008499ea8adda475de1042d6975c7b # v6 + with: + role-to-assume: ${{ vars.GH_BENCH_INGEST_ROLE_ARN }} + aws-region: ${{ vars.RDS_BENCH_REGION }} + - name: Install uv for v4 ingest + if: vars.GH_BENCH_INGEST_ROLE_ARN != '' + continue-on-error: true + uses: spiraldb/actions/.github/actions/setup-uv@a746510eafaa926484c354541cfc49b2ec06cc63 # 0.18.6 + - name: Ingest results to v4 Postgres (best-effort) + if: vars.GH_BENCH_INGEST_ROLE_ARN != '' + continue-on-error: true + shell: bash + env: + RDS_BENCH_INSTANCE_ENDPOINT: ${{ vars.RDS_BENCH_INSTANCE_ENDPOINT }} + RDS_BENCH_DB_NAME: ${{ vars.RDS_BENCH_DB_NAME }} + AWS_REGION: ${{ vars.RDS_BENCH_REGION }} + BENCH_SITE_BASE_URL: ${{ vars.BENCH_SITE_BASE_URL }} + BENCH_REVALIDATE_TOKEN: ${{ secrets.BENCH_REVALIDATE_TOKEN }} + run: | + set -Eeuo pipefail + curl -fsSL https://truststore.pki.rds.amazonaws.com/global/global-bundle.pem \ + -o "${RUNNER_TEMP}/rds-global-bundle.pem" + DSN="postgresql://bench_ingest@${RDS_BENCH_INSTANCE_ENDPOINT}:5432/${RDS_BENCH_DB_NAME}?sslmode=verify-full&sslrootcert=${RUNNER_TEMP}/rds-global-bundle.pem" + uv run --no-project --with 'psycopg[binary]' --with boto3 --with xxhash \ + scripts/post-ingest.py results.v3.jsonl \ + --postgres "${DSN}" \ + --commit-sha "${{ github.sha }}" \ + --region "${AWS_REGION}" + - name: Alert incident.io if: failure() uses: ./.github/actions/alert-incident-io diff --git a/.github/workflows/sql-benchmarks.yml b/.github/workflows/sql-benchmarks.yml index 3686934af39..ffefa28b802 100644 --- a/.github/workflows/sql-benchmarks.yml +++ b/.github/workflows/sql-benchmarks.yml @@ -682,6 +682,45 @@ jobs: --benchmark-id "${{ matrix.id }}" \ --repo-url "${{ github.server_url }}/${{ github.repository }}" + # v4 (Postgres) dual-write -- BEST-EFFORT during the migration soak (see bench.yml + # for the full rationale). The v3 step above is hard-required; a v4 failure must NOT + # fail the job (v4 is promoted to required at cutover, PR-5.1). Gated on + # inputs.mode == 'develop' (matching v3) + the ingest-role ARN var (the assume-role + # input that MUST exist for OIDC to succeed; it no-ops until v4 infra is wired), and + # every step is continue-on-error. post-ingest.py mints the RDS IAM token (boto3) from + # the assumed GitHubBenchmarkIngestRole; sslmode=verify-full validates the cert. + - name: Configure AWS credentials for v4 ingest (OIDC) + if: inputs.mode == 'develop' && vars.GH_BENCH_INGEST_ROLE_ARN != '' + continue-on-error: true + uses: aws-actions/configure-aws-credentials@e7f100cf4c008499ea8adda475de1042d6975c7b # v6 + with: + role-to-assume: ${{ vars.GH_BENCH_INGEST_ROLE_ARN }} + aws-region: ${{ vars.RDS_BENCH_REGION }} + - name: Install uv for v4 ingest + if: inputs.mode == 'develop' && vars.GH_BENCH_INGEST_ROLE_ARN != '' + continue-on-error: true + uses: spiraldb/actions/.github/actions/setup-uv@a746510eafaa926484c354541cfc49b2ec06cc63 # 0.18.6 + - name: Ingest results to v4 Postgres (best-effort) + if: inputs.mode == 'develop' && vars.GH_BENCH_INGEST_ROLE_ARN != '' + continue-on-error: true + shell: bash + env: + RDS_BENCH_INSTANCE_ENDPOINT: ${{ vars.RDS_BENCH_INSTANCE_ENDPOINT }} + RDS_BENCH_DB_NAME: ${{ vars.RDS_BENCH_DB_NAME }} + AWS_REGION: ${{ vars.RDS_BENCH_REGION }} + BENCH_SITE_BASE_URL: ${{ vars.BENCH_SITE_BASE_URL }} + BENCH_REVALIDATE_TOKEN: ${{ secrets.BENCH_REVALIDATE_TOKEN }} + run: | + set -Eeuo pipefail + curl -fsSL https://truststore.pki.rds.amazonaws.com/global/global-bundle.pem \ + -o "${RUNNER_TEMP}/rds-global-bundle.pem" + DSN="postgresql://bench_ingest@${RDS_BENCH_INSTANCE_ENDPOINT}:5432/${RDS_BENCH_DB_NAME}?sslmode=verify-full&sslrootcert=${RUNNER_TEMP}/rds-global-bundle.pem" + uv run --no-project --with 'psycopg[binary]' --with boto3 --with xxhash \ + scripts/post-ingest.py results.v3.jsonl \ + --postgres "${DSN}" \ + --commit-sha "${{ github.sha }}" \ + --region "${AWS_REGION}" + - name: Alert incident.io if: failure() && inputs.mode == 'develop' uses: ./.github/actions/alert-incident-io diff --git a/.github/workflows/v3-commit-metadata.yml b/.github/workflows/v3-commit-metadata.yml index d95da69207b..0e411730db8 100644 --- a/.github/workflows/v3-commit-metadata.yml +++ b/.github/workflows/v3-commit-metadata.yml @@ -9,6 +9,7 @@ on: workflow_dispatch: { } permissions: + id-token: write # enables AWS-GitHub OIDC for the best-effort v4 ingest step contents: read jobs: @@ -32,3 +33,41 @@ jobs: --commit-sha "${{ github.sha }}" \ --benchmark-id "commit-metadata" \ --repo-url "${{ github.server_url }}/${{ github.repository }}" + + # v4 (Postgres) dual-write -- BEST-EFFORT (see bench.yml rationale). Empty records: + # post-ingest.py --postgres upserts the commit row only. v3 above stays required; + # a v4 failure never fails the job (promoted to required at cutover, PR-5.1). + # Gated on the ingest-role ARN var (the assume-role input that MUST exist) so + # it no-ops until v4 infra is wired. + - name: Configure AWS credentials for v4 ingest (OIDC) + if: vars.GH_BENCH_INGEST_ROLE_ARN != '' + continue-on-error: true + uses: aws-actions/configure-aws-credentials@e7f100cf4c008499ea8adda475de1042d6975c7b # v6 + with: + role-to-assume: ${{ vars.GH_BENCH_INGEST_ROLE_ARN }} + aws-region: ${{ vars.RDS_BENCH_REGION }} + - name: Install uv for v4 ingest + if: vars.GH_BENCH_INGEST_ROLE_ARN != '' + continue-on-error: true + uses: spiraldb/actions/.github/actions/setup-uv@a746510eafaa926484c354541cfc49b2ec06cc63 # 0.18.6 + - name: Ingest commit metadata to v4 Postgres (best-effort) + if: vars.GH_BENCH_INGEST_ROLE_ARN != '' + continue-on-error: true + shell: bash + env: + RDS_BENCH_INSTANCE_ENDPOINT: ${{ vars.RDS_BENCH_INSTANCE_ENDPOINT }} + RDS_BENCH_DB_NAME: ${{ vars.RDS_BENCH_DB_NAME }} + AWS_REGION: ${{ vars.RDS_BENCH_REGION }} + BENCH_SITE_BASE_URL: ${{ vars.BENCH_SITE_BASE_URL }} + BENCH_REVALIDATE_TOKEN: ${{ secrets.BENCH_REVALIDATE_TOKEN }} + run: | + set -Eeuo pipefail + echo -n > empty.jsonl + curl -fsSL https://truststore.pki.rds.amazonaws.com/global/global-bundle.pem \ + -o "${RUNNER_TEMP}/rds-global-bundle.pem" + DSN="postgresql://bench_ingest@${RDS_BENCH_INSTANCE_ENDPOINT}:5432/${RDS_BENCH_DB_NAME}?sslmode=verify-full&sslrootcert=${RUNNER_TEMP}/rds-global-bundle.pem" + uv run --no-project --with 'psycopg[binary]' --with boto3 --with xxhash \ + scripts/post-ingest.py empty.jsonl \ + --postgres "${DSN}" \ + --commit-sha "${{ github.sha }}" \ + --region "${AWS_REGION}"