From b09f54576309c6216249b2c860bd0adaf58ab2c8 Mon Sep 17 00:00:00 2001 From: Maxwell Elliott Date: Wed, 29 Jul 2026 13:11:20 -0400 Subject: [PATCH] test: add cross-instance hash-consistency harness for `bazel-diff serve` Nothing today can catch a `serve` fleet disagreeing about a commit's hashes. `E2ETest#testGenerateHashesIsHermeticAcrossWorkspacePaths` runs two hashes in one JVM on one machine, and `serve_harness.py` drives a single instance. If target hashes carry host-specific state, replicas return different impacted targets and nothing notices. `tools/serve_consistency.py` runs N instances against one mock S3 cache tier and compares what they each wrote. The S3 tier is the observation point because `HashService.generate` persists the full `{label: hash}` map per revision, so a failure names the diverging targets rather than just reporting that the impacted sets differed. It also gives `S3HashCacheStorage` / `TieredHashCacheStorage` their first live-wire coverage. Each instance writes under its own `--s3Prefix` so writes stay attributable; `--s3Prefix` is absent from `computeConfigFingerprint()`, so they still agree on the cache key. Skew is injected only outside that fingerprint (clone paths of differing length, hence differing Bazel output bases; optionally HOME/TMPDIR/USER) -- skewing a fingerprinted flag would split the fleet across cache keys and leave nothing to compare. Cases: baseline (independent generation, payloads must agree), shared (followers must be served by the leader's objects over S3), concurrent (racing writers must not disagree), negative-control (one instance seeded differently *must* be detected, so a green run means something), plus opt-in version-skew and real-repository modes. Details that took some finding: - PutObject arrives aws-chunked. The SDK pins CHUNK_ENCODING_ENABLED for PutObject and the default CRC32 trailer turns framing on, so a naive body read records framed bytes rather than JSON. `mock_s3.decode_aws_chunked` handles the signed and unsigned spellings. - `--s3Region` is mandatory: the region resolves eagerly in `createS3Storage()` with no try/catch, so omitting it kills the process at startup. Credentials are separately mandatory because `S3HashCacheStorage` swallows every `SdkException` into a silent miss -- which is also why the harness asserts the mock actually received writes. - Byte-identity is the wrong gate. `HashService.serialize` gsons a HashMap built by `Collectors.toMap` over a parallel stream, so JSON key order tracks the host's CPU count. The comparator is semantic and reports key-order-only differences as a pass. `serve_harness.serve()` gains defaulted `extra_args` / `env` / `port` / `bazel` kwargs plus a `reserve_ports()` helper; existing behaviour is unchanged and `serve_stress.py` benefits too. Verified: hermetic matrix 47 pass / 0 fail; against this repository at b7ee642..a4c50ac, three instances on three output bases produced byte-identical payloads for all 319 targets and identical impacted sets. The workflow is cron-only for now, matching serve-harness.yml's precedent -- three concurrent Bazel servers is too heavy to gate every PR on until it has a track record. Co-Authored-By: Claude Opus 5 --- .github/workflows/serve-consistency.yml | 63 ++ tools/BUILD | 47 +- tools/mock_s3.py | 639 +++++++++++ tools/mock_s3_test.py | 261 +++++ tools/serve_consistency.py | 1324 +++++++++++++++++++++++ tools/serve_consistency_test.py | 360 ++++++ tools/serve_harness.py | 49 +- 7 files changed, 2736 insertions(+), 7 deletions(-) create mode 100644 .github/workflows/serve-consistency.yml create mode 100644 tools/mock_s3.py create mode 100644 tools/mock_s3_test.py create mode 100644 tools/serve_consistency.py create mode 100644 tools/serve_consistency_test.py diff --git a/.github/workflows/serve-consistency.yml b/.github/workflows/serve-consistency.yml new file mode 100644 index 00000000..786a3b22 --- /dev/null +++ b/.github/workflows/serve-consistency.yml @@ -0,0 +1,63 @@ +name: Serve Consistency + +# Cross-instance hash-consistency check for the `bazel-diff serve` query service. Runs +# tools/serve_consistency.py, which starts several real `serve` processes against one mock S3 cache +# tier and verifies they compute identical hashes for the same commit. A failure means target +# hashes carry host-specific state, which makes every impacted-target answer the fleet gives +# suspect. See the harness module docstring for the case matrix. +# +# Cron-only for now, matching serve-harness.yml's precedent: several concurrent Bazel servers per +# run is too heavy to gate every PR on until the harness has a healthy track record. +# `workflow_dispatch` allows on-demand runs from the Actions tab in the meantime. Note that +# scheduled + dispatched workflows run from the default branch, so this must land on master before +# the cron fires. +on: + workflow_dispatch: + schedule: + - cron: "30 3 * * *" + +jobs: + Consistency: + runs-on: ubuntu-latest + # Three instances means three independent Bazel output bases and three Bazel servers, so this + # is meaningfully slower than the single-instance serve harness. + timeout-minutes: 45 + strategy: + fail-fast: false + matrix: + bazel: [ '8.x', '9.x' ] + steps: + - name: Setup Java JDK + uses: actions/setup-java@v4 + with: + distribution: 'temurin' + java-version: '21' + - name: Setup Go environment + uses: actions/setup-go@v5 + with: + go-version: ^1.17 + id: go + - name: Setup Bazelisk + run: go install github.com/bazelbuild/bazelisk@latest + - uses: actions/checkout@v4 + - name: Run consistency harness + env: + USE_BAZEL_VERSION: ${{ matrix.bazel }} + # git daemon ships with git (preinstalled on the runner); python3 is preinstalled. The mock + # S3 backend is pure stdlib, so no service container is needed. + run: | + python3 tools/serve_consistency.py \ + --bazel "$HOME/go/bin/bazelisk" \ + --instances 3 \ + --metrics-out "serve-consistency-${{ matrix.bazel }}-metrics.json" \ + --summary-out "serve-consistency-${{ matrix.bazel }}-summary.md" + - name: Publish summary + if: always() + run: cat "serve-consistency-${{ matrix.bazel }}-summary.md" >> "$GITHUB_STEP_SUMMARY" || true + - name: Upload metrics + if: always() + uses: actions/upload-artifact@v4 + with: + name: serve-consistency-metrics-${{ matrix.bazel }} + path: serve-consistency-${{ matrix.bazel }}-metrics.json + if-no-files-found: ignore diff --git a/tools/BUILD b/tools/BUILD index 2eb67638..ba672156 100644 --- a/tools/BUILD +++ b/tools/BUILD @@ -55,11 +55,12 @@ py_test( deps = [":coverage_check_lib"], ) -# Note: the serve integration harnesses (serve_harness.py, serve_stress.py) are intentionally NOT -# wrapped in py_binary targets. They shell out to `bazel build //cli:bazel-diff`, which would -# deadlock on the output-base lock if they were themselves launched via `bazel run` (nested bazel, -# same output base). Run them directly: `python3 tools/serve_harness.py` / -# `python3 tools/serve_stress.py` (see their module docstrings). They use only the Python stdlib. +# Note: the serve integration harnesses (serve_harness.py, serve_stress.py, serve_consistency.py) +# are intentionally NOT wrapped in py_binary targets. They shell out to +# `bazel build //cli:bazel-diff`, which would deadlock on the output-base lock if they were +# themselves launched via `bazel run` (nested bazel, same output base). Run them directly: +# `python3 tools/serve_harness.py` / `python3 tools/serve_stress.py` / +# `python3 tools/serve_consistency.py` (see their module docstrings). Python stdlib only. # # A py_test over their *pure* helpers is fine, though -- it imports the module and calls functions # that never shell out, so no nested bazel and no deadlock. serve_stress_test.py covers the CLI @@ -81,3 +82,39 @@ py_test( python_version = "PY3", deps = [":serve_stress_lib"], ) + +# mock_s3.py is the one harness-adjacent module that never invokes bazel or git, so its own test +# can drive the live HTTP server as well as the pure helpers. +py_library( + name = "mock_s3_lib", + srcs = ["mock_s3.py"], + imports = ["."], + visibility = ["//visibility:private"], +) + +py_test( + name = "mock_s3_test", + srcs = ["mock_s3_test.py"], + python_version = "PY3", + deps = [":mock_s3_lib"], +) + +# serve_consistency_test.py covers the payload comparator that decides whether a fleet's hashes +# agree; a false negative there would report consistency that was never actually checked. +py_library( + name = "serve_consistency_lib", + srcs = [ + "mock_s3.py", + "serve_consistency.py", + "serve_harness.py", # imported by serve_consistency as `base` + ], + imports = ["."], + visibility = ["//visibility:private"], +) + +py_test( + name = "serve_consistency_test", + srcs = ["serve_consistency_test.py"], + python_version = "PY3", + deps = [":serve_consistency_lib"], +) diff --git a/tools/mock_s3.py b/tools/mock_s3.py new file mode 100644 index 00000000..7c93ef03 --- /dev/null +++ b/tools/mock_s3.py @@ -0,0 +1,639 @@ +#!/usr/bin/env python3 +"""A recording, in-memory mock of the small slice of S3 that `bazel-diff serve` uses. + +`S3HashCacheStorage` only ever issues GetObject, PutObject and HeadObject against a single bucket +(see cli/src/main/kotlin/com/bazel_diff/server/S3HashCacheStorage.kt), so a few hundred lines of +stdlib `http.server` stand in for MinIO/LocalStack with no container and no third-party dependency +-- and, unlike a real bucket, this one *records every request*. That recording is the point: it is +what lets `serve_consistency.py` compare the hash payloads several `serve` instances independently +produced for the same revision, and prove which instance served which read. + +Two read modes, chosen per run: + + * "isolated" -- a GET/HEAD only resolves objects written under the *requesting* instance's own + key prefix. Every instance therefore generates its own hashes for a revision, giving us N + independent payloads under one cache key to diff. This is the hash-poisoning detector. + * "shared" -- a GET/HEAD resolves against one logical namespace regardless of prefix, i.e. the + real shared-cache topology. `ObjectRead.served_from_prefix` still records *whose* write + satisfied the read, which is what turns a `cacheHit: true` in a serve profile into proof that + the remote tier (rather than the instance's own local disk tier) served it. + +Instances always write under their own prefix, so writes stay attributable in both modes. + +Wire notes (AWS SDK v2 2.46.7, the version pinned in MODULE.bazel): + + * PutObject arrives **`aws-chunked`-framed**. The SDK's S3 auth-scheme defaults pin + `PAYLOAD_SIGNING_ENABLED=false` but `CHUNK_ENCODING_ENABLED=true` for PutObject, and the + default `RequestChecksumCalculation=WHEN_SUPPORTED` appends a CRC32 *trailer* -- which turns + chunk encoding on. A naive `read(Content-Length)` therefore stores framed bytes, not the JSON, + and every payload comparison downstream would be garbage. `decode_aws_chunked` handles both + the unsigned (`\\r\\n`) and signed (`;chunk-signature=\\r\\n`) spellings. + * PutObject sends `Expect: 100-continue`; `protocol_version = "HTTP/1.1"` makes + `BaseHTTPRequestHandler` answer the handshake automatically. + * A missing key must come back as `404` with an `NoSuchKey` body for GET (the + SDK maps that to `NoSuchKeyException`, the one exception `S3HashCacheStorage.get` treats as a + plain miss rather than a warning) and as a *bare* `404` for HEAD. + * Never return 5xx. The SDK retries those, which would double-count writes and silently mask a + mock bug as a cache miss. Unhandled requests are recorded in `bad_requests()` and answered + `400`, so a harness can assert the list is empty instead of going green on nothing. + +Pure stdlib. `decode_aws_chunked`, `parse_object_path`, `split_prefix` and `error_xml` are free +functions with no I/O so they can be unit-tested directly (see mock_s3_test.py). +""" + +from __future__ import annotations + +import contextlib +import email.utils +import hashlib +import threading +import time +import urllib.parse +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Iterable, Mapping, Sequence + +# ---------------------------------------------------------------------------------------------- +# Pure helpers +# ---------------------------------------------------------------------------------------------- + + +class ChunkDecodeError(Exception): + """Raised when an `aws-chunked` body is malformed or its decoded length is not as declared.""" + + +def decode_aws_chunked(body: bytes, expected_len: int | None = None) -> tuple[bytes, dict]: + """Strips `aws-chunked` framing, returning (payload, trailing_headers). + + The framing is a sequence of `[;chunk-signature=]\\r\\n\\r\\n` chunks + terminated by a zero-length chunk, optionally followed by trailer headers (the SDK sends + `x-amz-checksum-crc32` there) and a blank line. Signed and unsigned chunk headers differ only + in the `;`-suffixed parameters, which we discard -- there is nothing to verify against, since + the mock ignores SigV4 entirely. + + [expected_len], when given, is the `x-amz-decoded-content-length` header; a mismatch means we + mis-parsed the stream and must fail loudly rather than record a corrupt payload. + """ + payload = bytearray() + trailers: dict = {} + pos = 0 + end = len(body) + + while True: + eol = body.find(b"\r\n", pos) + if eol == -1: + raise ChunkDecodeError(f"unterminated chunk header at offset {pos}") + size_field = body[pos:eol].split(b";", 1)[0].strip() + pos = eol + 2 + try: + size = int(size_field, 16) + except ValueError as exc: + raise ChunkDecodeError(f"bad chunk size {size_field!r} at offset {pos}") from exc + if size == 0: + break + if pos + size > end: + raise ChunkDecodeError(f"chunk of {size} bytes overruns the body at offset {pos}") + payload += body[pos : pos + size] + pos += size + if body[pos : pos + 2] != b"\r\n": + raise ChunkDecodeError(f"missing CRLF after chunk data at offset {pos}") + pos += 2 + + # Whatever follows the terminating chunk is trailer headers, ending at a blank line. + while pos < end: + eol = body.find(b"\r\n", pos) + line = body[pos:end] if eol == -1 else body[pos:eol] + pos = end if eol == -1 else eol + 2 + if not line.strip(): + continue + if b":" in line: + name, value = line.split(b":", 1) + trailers[name.decode("utf-8", "replace").strip().lower()] = value.decode( + "utf-8", "replace" + ).strip() + + decoded = bytes(payload) + if expected_len is not None and len(decoded) != expected_len: + raise ChunkDecodeError( + f"decoded {len(decoded)} bytes but x-amz-decoded-content-length said {expected_len}" + ) + return decoded, trailers + + +def parse_object_path(path: str) -> tuple[str, str] | None: + """`/bucket/a/b.json?x=1` -> `("bucket", "a/b.json")`, percent-decoded. + + Returns None for anything that is not a path-style object address (the service root, a bucket + listing, a malformed path). Path style is what `--s3ForcePathStyle` produces, and it is the + only addressing this mock accepts -- virtual-host style would need wildcard DNS. + """ + raw = path.split("?", 1)[0] + if not raw.startswith("/"): + return None + parts = raw[1:].split("/", 1) + if len(parts) != 2 or not parts[0] or not parts[1]: + return None + return urllib.parse.unquote(parts[0]), urllib.parse.unquote(parts[1]) + + +def split_prefix(key: str, prefixes: Iterable[str]) -> tuple[str | None, str]: + """Splits a full object key into (owning_prefix, logical_key). + + Longest prefix wins, so `i1/` and `i11/` cannot be confused. A key under no known prefix comes + back as `(None, key)` -- the caller records it rather than guessing. + """ + best: str | None = None + for prefix in prefixes: + if key.startswith(prefix) and (best is None or len(prefix) > len(best)): + best = prefix + if best is None: + return None, key + return best, key[len(best) :] + + +def error_xml(code: str, message: str, key: str = "") -> bytes: + """The S3 REST error document. `` is the field the SDK maps to an exception type.""" + return ( + '' + "" + f"{code}" + f"{message}" + f"{key}" + "mock-s3" + "mock-s3" + "" + ).encode("utf-8") + + +# ---------------------------------------------------------------------------------------------- +# Recorded traffic +# ---------------------------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class ObjectWrite: + seq: int + ts: float + instance: str | None # instance id that owns `prefix`, None if the prefix is unregistered + prefix: str + logical_key: str # the cache key with the instance prefix stripped: `.` + body: bytes + sha256: str + size: int + chunked: bool # whether the request arrived aws-chunked (it should) + + +@dataclass(frozen=True) +class ObjectRead: + seq: int + ts: float + instance: str | None + prefix: str + logical_key: str + op: str # "GET" | "HEAD" + hit: bool + served_from_prefix: str | None # in shared mode, whose write satisfied this read + + +# ---------------------------------------------------------------------------------------------- +# The store +# ---------------------------------------------------------------------------------------------- + +ISOLATED = "isolated" +SHARED = "shared" + + +class MockS3Store: + """Thread-safe object store plus request log. Shared by every handler thread.""" + + def __init__( + self, + bucket: str, + prefixes: Mapping[str, str] | None = None, + mode: str = ISOLATED, + ) -> None: + if mode not in (ISOLATED, SHARED): + raise ValueError(f"unknown mode {mode!r}") + self.bucket = bucket + self._mode = mode + self._lock = threading.Lock() + self._prefix_to_instance: dict = {} + self._objects: dict = {} # full key -> bytes + self._writes: list = [] + self._reads: list = [] + self._bad: list = [] + self._seq = 0 + # Traffic from prior, already-reset generations, so a caller that resets between phases can + # still report what the whole run did. + self._retired = {"puts": 0, "gets": 0, "heads": 0, "hits": 0, "bad_requests": 0} + for iid, prefix in (prefixes or {}).items(): + self.register(iid, prefix) + + # -- configuration --------------------------------------------------------------------- + + def register(self, instance: str, prefix: str) -> None: + """Declares that [prefix] belongs to [instance], so its traffic is attributable.""" + with self._lock: + self._prefix_to_instance[prefix] = instance + + @property + def mode(self) -> str: + return self._mode + + def set_mode(self, mode: str) -> None: + if mode not in (ISOLATED, SHARED): + raise ValueError(f"unknown mode {mode!r}") + with self._lock: + self._mode = mode + + def reset(self) -> None: + """Drops every object and every recorded request. Cases call this at their boundary so one + case's writes cannot make the next case's "the follower read from i0" assertion vacuous. + + The request *counts* survive in a retired tally so `total_counts()` still describes the + whole run; only the objects and the per-request log are cleared. + """ + with self._lock: + current = self._counts_locked() + for key in self._retired: + self._retired[key] += current[key] + self._objects.clear() + self._writes.clear() + self._reads.clear() + self._bad.clear() + self._seq = 0 + + # -- operations ------------------------------------------------------------------------ + + def _next_seq(self) -> int: + self._seq += 1 + return self._seq + + def _attribute(self, full_key: str) -> tuple[str | None, str, str]: + prefix, logical = split_prefix(full_key, self._prefix_to_instance) + if prefix is None: + return None, "", full_key + return self._prefix_to_instance[prefix], prefix, logical + + def _resolve(self, full_key: str) -> tuple[bytes | None, str | None]: + """Finds the bytes for [full_key], honouring the read mode. Returns (body, served_from).""" + own = self._objects.get(full_key) + if own is not None: + _, prefix, _ = self._attribute(full_key) + return own, prefix + if self._mode != SHARED: + return None, None + _, _, logical = self._attribute(full_key) + # Earliest writer wins, so a follower's read is attributed to whoever actually published. + for write in self._writes: + if write.logical_key == logical: + stored = self._objects.get(write.prefix + logical) + if stored is not None: + return stored, write.prefix + return None, None + + def put(self, full_key: str, body: bytes, chunked: bool) -> None: + with self._lock: + instance, prefix, logical = self._attribute(full_key) + self._objects[full_key] = body + self._writes.append( + ObjectWrite( + seq=self._next_seq(), + ts=time.time(), + instance=instance, + prefix=prefix, + logical_key=logical, + body=body, + sha256=hashlib.sha256(body).hexdigest(), + size=len(body), + chunked=chunked, + ) + ) + + def get(self, full_key: str) -> bytes | None: + with self._lock: + body, served_from = self._resolve(full_key) + self._record_read(full_key, "GET", body is not None, served_from) + return body + + def head(self, full_key: str) -> int | None: + with self._lock: + body, served_from = self._resolve(full_key) + self._record_read(full_key, "HEAD", body is not None, served_from) + return None if body is None else len(body) + + def _record_read(self, full_key, op, hit, served_from) -> None: + """Caller must hold the lock.""" + instance, prefix, logical = self._attribute(full_key) + self._reads.append( + ObjectRead( + seq=self._next_seq(), + ts=time.time(), + instance=instance, + prefix=prefix, + logical_key=logical, + op=op, + hit=hit, + served_from_prefix=served_from, + ) + ) + + def record_bad_request(self, method: str, path: str, detail: str = "") -> None: + with self._lock: + self._bad.append((method, path, detail)) + + # -- inspection ------------------------------------------------------------------------ + + def writes(self, *, instance: str | None = None, logical_key: str | None = None) -> list: + with self._lock: + return [ + w + for w in self._writes + if (instance is None or w.instance == instance) + and (logical_key is None or w.logical_key == logical_key) + ] + + def reads( + self, + *, + instance: str | None = None, + op: str | None = None, + logical_key: str | None = None, + ) -> list: + with self._lock: + return [ + r + for r in self._reads + if (instance is None or r.instance == instance) + and (op is None or r.op == op) + and (logical_key is None or r.logical_key == logical_key) + ] + + def logical_keys(self) -> list: + """Every cache key observed, in first-write order.""" + with self._lock: + seen: list = [] + for w in self._writes: + if w.logical_key not in seen: + seen.append(w.logical_key) + return seen + + def bodies_for(self, logical_key: str) -> dict: + """instance id -> the payload it wrote for [logical_key] (its last write, if it wrote more + than once). Exactly the input `serve_consistency.compare_payloads` expects.""" + with self._lock: + out: dict = {} + for w in self._writes: + if w.logical_key == logical_key and w.instance is not None: + out[w.instance] = w.body + return out + + def bad_requests(self) -> list: + with self._lock: + return list(self._bad) + + def _counts_locked(self) -> dict: + return { + "puts": len(self._writes), + "gets": sum(1 for r in self._reads if r.op == "GET"), + "heads": sum(1 for r in self._reads if r.op == "HEAD"), + "hits": sum(1 for r in self._reads if r.hit), + "objects": len(self._objects), + "bad_requests": len(self._bad), + } + + def counts(self) -> dict: + """Traffic since the last `reset()`.""" + with self._lock: + return self._counts_locked() + + def total_counts(self) -> dict: + """Traffic across the whole run, including generations already cleared by `reset()`.""" + with self._lock: + out = self._counts_locked() + for key, value in self._retired.items(): + out[key] += value + return out + + def counts_by_instance(self) -> dict: + with self._lock: + out: dict = {} + for iid in set(self._prefix_to_instance.values()): + out[iid] = {"puts": 0, "gets": 0, "heads": 0, "hits": 0} + for w in self._writes: + if w.instance in out: + out[w.instance]["puts"] += 1 + for r in self._reads: + if r.instance in out: + out[r.instance]["gets" if r.op == "GET" else "heads"] += 1 + if r.hit: + out[r.instance]["hits"] += 1 + return out + + +# ---------------------------------------------------------------------------------------------- +# HTTP surface +# ---------------------------------------------------------------------------------------------- + + +class _Handler(BaseHTTPRequestHandler): + # HTTP/1.1 both enables keep-alive (the SDK reuses connections) and makes the base class + # answer the `Expect: 100-continue` PutObject sends. + protocol_version = "HTTP/1.1" + server_version = "MockS3/1.0" + + # Injected by MockS3Server. + store: MockS3Store + + def log_message(self, fmt, *args) -> None: # noqa: A003 - stdlib signature + """Silenced; the store is the log.""" + + # -- helpers --------------------------------------------------------------------------- + + def _object_key(self) -> str | None: + parsed = parse_object_path(self.path) + if parsed is None: + return None + bucket, key = parsed + if bucket != self.store.bucket: + return None + return key + + def _send(self, code: int, body: bytes = b"", ctype: str | None = None, extra=None) -> None: + self.send_response(code) + if ctype: + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(body))) + self.send_header("x-amz-request-id", "mock-s3") + for name, value in (extra or {}).items(): + self.send_header(name, value) + self.end_headers() + if body: + self.wfile.write(body) + + def _send_headers_only(self, code: int, content_length: int, extra=None) -> None: + """A HEAD response: the same headers a GET would carry, but never a body.""" + self.send_response(code) + self.send_header("Content-Length", str(content_length)) + self.send_header("x-amz-request-id", "mock-s3") + for name, value in (extra or {}).items(): + self.send_header(name, value) + self.end_headers() + + def _no_such_key(self, key: str, head: bool) -> None: + if head: + # HeadObject reports a missing key as a bare 404 with no error body; that is exactly + # the shape S3HashCacheStorage.contains() special-cases. + self._send_headers_only(404, 0) + else: + self._send(404, error_xml("NoSuchKey", "The specified key does not exist.", key), + "application/xml") + + def _read_body(self) -> tuple[bytes, bool]: + """Reads the request body, undoing `aws-chunked` framing. Returns (payload, was_chunked).""" + encoding = (self.headers.get("Content-Encoding") or "").lower() + decoded_len = self.headers.get("x-amz-decoded-content-length") + if (self.headers.get("Transfer-Encoding") or "").lower() == "chunked": + raw = self._read_http_chunked() + else: + raw = self.rfile.read(int(self.headers.get("Content-Length") or 0)) + if "aws-chunked" in encoding or decoded_len is not None: + expected = int(decoded_len) if decoded_len is not None else None + return decode_aws_chunked(raw, expected)[0], True + return raw, False + + def _read_http_chunked(self) -> bytes: + """Transport-level `Transfer-Encoding: chunked`, which the SDK does not currently use for + PutObject but which costs little to tolerate.""" + out = bytearray() + while True: + line = self.rfile.readline().split(b";", 1)[0].strip() + size = int(line, 16) + if size == 0: + # Consume trailers up to the blank line. + while self.rfile.readline().strip(): + pass + break + out += self.rfile.read(size) + self.rfile.read(2) + return bytes(out) + + # -- verbs ----------------------------------------------------------------------------- + + def do_PUT(self) -> None: # noqa: N802 - stdlib naming + key = self._object_key() + if key is None: + self.store.record_bad_request("PUT", self.path, "unroutable object path") + self._send(400, error_xml("InvalidRequest", "unroutable path"), "application/xml") + return + try: + body, chunked = self._read_body() + except (ChunkDecodeError, ValueError) as exc: + # Deliberately a 4xx: a 5xx would be retried by the SDK and the resulting cache miss + # would look like "no divergence found" instead of "the mock is broken". + self.store.record_bad_request("PUT", self.path, f"body decode failed: {exc}") + self._send(400, error_xml("InvalidRequest", str(exc), key), "application/xml") + return + self.store.put(key, body, chunked) + self._send(200, b"", extra={"ETag": '"%s"' % hashlib.md5(body).hexdigest()}) + + def do_GET(self) -> None: # noqa: N802 + key = self._object_key() + if key is None: + self.store.record_bad_request("GET", self.path, "unroutable object path") + self._send(400, error_xml("InvalidRequest", "unroutable path"), "application/xml") + return + body = self.store.get(key) + if body is None: + self._no_such_key(key, head=False) + return + self._send( + 200, + body, + "application/json", + extra={ + "ETag": '"%s"' % hashlib.md5(body).hexdigest(), + "Last-Modified": email.utils.formatdate(usegmt=True), + }, + ) + + def do_HEAD(self) -> None: # noqa: N802 + key = self._object_key() + if key is None: + self.store.record_bad_request("HEAD", self.path, "unroutable object path") + self._send_headers_only(400, 0) + return + size = self.store.head(key) + if size is None: + self._no_such_key(key, head=True) + return + self._send_headers_only(200, size, extra={"Content-Type": "application/json"}) + + def do_DELETE(self) -> None: # noqa: N802 + # serve never deletes; answer politely and record it so an unexpected op is visible. + self.store.record_bad_request("DELETE", self.path, "unexpected operation") + self._send(204) + + def do_POST(self) -> None: # noqa: N802 + self.store.record_bad_request("POST", self.path, "unexpected operation") + self._send(400, error_xml("NotImplemented", "unsupported operation"), "application/xml") + + +class MockS3Server: + """Runs [store] on a loopback HTTP port. Use as a context manager or call start()/stop().""" + + def __init__(self, store: MockS3Store, host: str = "127.0.0.1", port: int = 0) -> None: + self.store = store + handler = type("_BoundHandler", (_Handler,), {"store": store}) + self._httpd = ThreadingHTTPServer((host, port), handler) + self._httpd.daemon_threads = True + self._thread: threading.Thread | None = None + self.host = host + # Read the port back off the live socket rather than pre-allocating one, so there is no + # window between choosing a port and binding it. + self.port = self._httpd.server_address[1] + + @property + def url(self) -> str: + return f"http://{self.host}:{self.port}" + + def start(self) -> None: + self._thread = threading.Thread(target=self._httpd.serve_forever, daemon=True) + self._thread.start() + + def stop(self) -> None: + with contextlib.suppress(Exception): + self._httpd.shutdown() + with contextlib.suppress(Exception): + self._httpd.server_close() + if self._thread is not None: + self._thread.join(timeout=5) + + def __enter__(self) -> "MockS3Server": + self.start() + return self + + def __exit__(self, *exc) -> None: + self.stop() + + +@contextlib.contextmanager +def mock_s3(bucket: str, prefixes: Mapping[str, str] | None = None, mode: str = ISOLATED): + """Yields a started (MockS3Store, MockS3Server) pair and shuts the server down on exit.""" + store = MockS3Store(bucket, prefixes, mode) + server = MockS3Server(store) + server.start() + try: + yield store, server + finally: + server.stop() + + +def instance_prefixes(instance_ids: Sequence[str]) -> dict: + """`["i0", "i1"]` -> `{"i0": "i0/", "i1": "i1/"}` -- the prefix layout the harness uses. + + Note that `--s3Prefix` is deliberately absent from `ServeCommand.computeConfigFingerprint()`, + so giving each instance its own prefix keeps writes attributable *without* changing the cache + key the instances compute. That is what makes cross-instance comparison possible at all. + """ + return {iid: f"{iid}/" for iid in instance_ids} diff --git a/tools/mock_s3_test.py b/tools/mock_s3_test.py new file mode 100644 index 00000000..835543b1 --- /dev/null +++ b/tools/mock_s3_test.py @@ -0,0 +1,261 @@ +#!/usr/bin/env python3 +"""Unit tests for tools/mock_s3.py. + +`mock_s3` is pure stdlib and never shells out to bazel or git, so unlike the serve harnesses it is +safe to exercise fully -- including its HTTP surface -- from a `py_test`. That matters because the +mock is load-bearing: if its `aws-chunked` decoder silently mis-parses a PutObject, the consistency +harness records corrupt payloads and reports divergence that is not real (or, worse, records +identical corruption and reports agreement that is not real). +""" + +import json +import unittest +import urllib.error +import urllib.request + +import mock_s3 + + +def _chunk(payload: bytes, *, signature: str | None = None) -> bytes: + header = f"{len(payload):x}" + if signature: + header += f";chunk-signature={signature}" + return header.encode() + b"\r\n" + payload + b"\r\n" + + +def _terminator(*, signature: str | None = None, trailers=()) -> bytes: + header = "0" + if signature: + header += f";chunk-signature={signature}" + out = header.encode() + b"\r\n" + for name, value in trailers: + out += f"{name}:{value}\r\n".encode() + out += b"\r\n" + return out + + +class DecodeAwsChunkedTest(unittest.TestCase): + def test_unsigned_single_chunk(self): + framed = _chunk(b'{"a":1}') + _terminator() + payload, trailers = mock_s3.decode_aws_chunked(framed) + self.assertEqual(payload, b'{"a":1}') + self.assertEqual(trailers, {}) + + def test_signed_chunks_are_accepted(self): + sig = "a" * 64 + framed = _chunk(b"hello", signature=sig) + _terminator(signature=sig) + payload, _ = mock_s3.decode_aws_chunked(framed) + self.assertEqual(payload, b"hello") + + def test_multiple_chunks_are_concatenated(self): + framed = _chunk(b"abc") + _chunk(b"defg") + _terminator() + payload, _ = mock_s3.decode_aws_chunked(framed) + self.assertEqual(payload, b"abcdefg") + + def test_trailers_are_captured(self): + framed = _chunk(b"x") + _terminator(trailers=[("x-amz-checksum-crc32", "Ag8Zpw==")]) + payload, trailers = mock_s3.decode_aws_chunked(framed) + self.assertEqual(payload, b"x") + self.assertEqual(trailers, {"x-amz-checksum-crc32": "Ag8Zpw=="}) + + def test_empty_payload(self): + payload, _ = mock_s3.decode_aws_chunked(_terminator()) + self.assertEqual(payload, b"") + + def test_declared_length_mismatch_raises(self): + framed = _chunk(b"abc") + _terminator() + with self.assertRaises(mock_s3.ChunkDecodeError): + mock_s3.decode_aws_chunked(framed, expected_len=99) + + def test_declared_length_match_passes(self): + framed = _chunk(b"abc") + _terminator() + payload, _ = mock_s3.decode_aws_chunked(framed, expected_len=3) + self.assertEqual(payload, b"abc") + + def test_truncated_chunk_raises(self): + with self.assertRaises(mock_s3.ChunkDecodeError): + mock_s3.decode_aws_chunked(b"10\r\nshort\r\n") + + def test_garbage_size_raises(self): + with self.assertRaises(mock_s3.ChunkDecodeError): + mock_s3.decode_aws_chunked(b"zz\r\ndata\r\n") + + +class ParseObjectPathTest(unittest.TestCase): + def test_simple(self): + self.assertEqual(mock_s3.parse_object_path("/bucket/key.json"), ("bucket", "key.json")) + + def test_nested_key_keeps_slashes(self): + self.assertEqual( + mock_s3.parse_object_path("/bucket/i0/abc.def.json"), ("bucket", "i0/abc.def.json") + ) + + def test_query_string_is_dropped(self): + self.assertEqual( + mock_s3.parse_object_path("/bucket/k.json?x-id=GetObject"), ("bucket", "k.json") + ) + + def test_percent_decoding(self): + self.assertEqual(mock_s3.parse_object_path("/bucket/a%20b.json"), ("bucket", "a b.json")) + + def test_non_object_paths(self): + self.assertIsNone(mock_s3.parse_object_path("/")) + self.assertIsNone(mock_s3.parse_object_path("/bucket")) + self.assertIsNone(mock_s3.parse_object_path("/bucket/")) + self.assertIsNone(mock_s3.parse_object_path("relative/key")) + + +class SplitPrefixTest(unittest.TestCase): + def test_longest_prefix_wins(self): + self.assertEqual(mock_s3.split_prefix("i11/k.json", ["i1/", "i11/"]), ("i11/", "k.json")) + + def test_unknown_prefix(self): + self.assertEqual(mock_s3.split_prefix("zz/k.json", ["i0/"]), (None, "zz/k.json")) + + +class ErrorXmlTest(unittest.TestCase): + def test_code_is_present(self): + body = mock_s3.error_xml("NoSuchKey", "The specified key does not exist.", "i0/k.json") + self.assertIn(b"NoSuchKey", body) + self.assertIn(b"i0/k.json", body) + + +# ---------------------------------------------------------------------------------------------- +# The live HTTP surface +# ---------------------------------------------------------------------------------------------- + + +def _request(url, method, data=None, headers=None): + req = urllib.request.Request(url, method=method, data=data, headers=headers or {}) + try: + with urllib.request.urlopen(req, timeout=10) as resp: + return resp.status, resp.read() + except urllib.error.HTTPError as e: + return e.code, e.read() + + +class ServerTest(unittest.TestCase): + BUCKET = "test-bucket" + + def setUp(self): + self.prefixes = mock_s3.instance_prefixes(["i0", "i1"]) + self.ctx = mock_s3.mock_s3(self.BUCKET, self.prefixes) + self.store, self.server = self.ctx.__enter__() + self.addCleanup(self.ctx.__exit__, None, None, None) + + def url(self, key): + return f"{self.server.url}/{self.BUCKET}/{key}" + + def test_put_then_get_roundtrip(self): + body = json.dumps({"//:core": "Rule#abc~def"}).encode() + code, _ = _request(self.url("i0/sha.fp.json"), "PUT", body) + self.assertEqual(code, 200) + code, got = _request(self.url("i0/sha.fp.json"), "GET") + self.assertEqual(code, 200) + self.assertEqual(got, body) + + def test_chunked_put_is_decoded(self): + payload = b'{"//:core":"Rule#a~b"}' + framed = _chunk(payload) + _terminator(trailers=[("x-amz-checksum-crc32", "AAAAAA==")]) + code, _ = _request( + self.url("i0/k.json"), + "PUT", + framed, + headers={ + "Content-Encoding": "aws-chunked", + "x-amz-decoded-content-length": str(len(payload)), + }, + ) + self.assertEqual(code, 200) + writes = self.store.writes(instance="i0") + self.assertEqual(len(writes), 1) + self.assertEqual(writes[0].body, payload) + self.assertTrue(writes[0].chunked) + + def test_undecodable_chunked_put_is_a_400_and_recorded(self): + code, _ = _request( + self.url("i0/k.json"), + "PUT", + b"not chunked at all", + headers={"Content-Encoding": "aws-chunked", "x-amz-decoded-content-length": "5"}, + ) + # Deliberately a 4xx: a 5xx would be retried by the SDK and look like a cache miss. + self.assertEqual(code, 400) + self.assertTrue(self.store.bad_requests()) + + def test_get_miss_is_no_such_key(self): + code, body = _request(self.url("i0/absent.json"), "GET") + self.assertEqual(code, 404) + self.assertIn(b"NoSuchKey", body) + + def test_head_hit_and_miss(self): + _request(self.url("i0/k.json"), "PUT", b"12345") + code, body = _request(self.url("i0/k.json"), "HEAD") + self.assertEqual(code, 200) + self.assertEqual(body, b"") + code, body = _request(self.url("i0/absent.json"), "HEAD") + self.assertEqual(code, 404) + self.assertEqual(body, b"") + + def test_isolated_mode_hides_other_instances_writes(self): + _request(self.url("i0/k.json"), "PUT", b"payload") + code, _ = _request(self.url("i1/k.json"), "GET") + self.assertEqual(code, 404) + reads = self.store.reads(instance="i1") + self.assertEqual(len(reads), 1) + self.assertFalse(reads[0].hit) + + def test_shared_mode_exposes_and_attributes_the_write(self): + _request(self.url("i0/k.json"), "PUT", b"payload") + self.store.set_mode(mock_s3.SHARED) + code, got = _request(self.url("i1/k.json"), "GET") + self.assertEqual(code, 200) + self.assertEqual(got, b"payload") + read = self.store.reads(instance="i1")[-1] + self.assertTrue(read.hit) + self.assertEqual(read.served_from_prefix, "i0/") + + def test_bodies_for_groups_writes_by_instance(self): + _request(self.url("i0/k.json"), "PUT", b"from-i0") + _request(self.url("i1/k.json"), "PUT", b"from-i1") + self.assertEqual(self.store.logical_keys(), ["k.json"]) + self.assertEqual( + self.store.bodies_for("k.json"), {"i0": b"from-i0", "i1": b"from-i1"} + ) + + def test_unknown_bucket_is_recorded_as_bad(self): + code, _ = _request(f"{self.server.url}/other-bucket/k.json", "GET") + self.assertEqual(code, 400) + self.assertTrue(self.store.bad_requests()) + + def test_counts_and_reset(self): + _request(self.url("i0/k.json"), "PUT", b"x") + _request(self.url("i0/k.json"), "GET") + counts = self.store.counts() + self.assertEqual(counts["puts"], 1) + self.assertEqual(counts["gets"], 1) + self.assertEqual(self.store.counts_by_instance()["i0"]["hits"], 1) + self.store.reset() + self.assertEqual(self.store.counts()["puts"], 0) + self.assertEqual(self.store.logical_keys(), []) + + def test_reset_retires_counts_into_the_run_total(self): + # The harness resets between cases; without this the reported traffic would describe only + # whichever case happened to run last. + _request(self.url("i0/k.json"), "PUT", b"x") + _request(self.url("i0/k.json"), "GET") + self.store.reset() + _request(self.url("i1/k.json"), "PUT", b"y") + self.assertEqual(self.store.counts()["puts"], 1) + total = self.store.total_counts() + self.assertEqual(total["puts"], 2) + self.assertEqual(total["gets"], 1) + + +class InstancePrefixesTest(unittest.TestCase): + def test_layout(self): + self.assertEqual(mock_s3.instance_prefixes(["i0", "i1"]), {"i0": "i0/", "i1": "i1/"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/serve_consistency.py b/tools/serve_consistency.py new file mode 100644 index 00000000..30a04d58 --- /dev/null +++ b/tools/serve_consistency.py @@ -0,0 +1,1324 @@ +#!/usr/bin/env python3 +"""Cross-instance hash-consistency harness for `bazel-diff serve`. + +Runs N `serve` instances side by side against one shared (mock) S3 cache tier and checks that they +produce **the same hashes for the same commit**. If they do not, target hashes are carrying +host-specific state, and every impacted-target answer the fleet gives is suspect. + +Nothing else in the repo can catch that. `E2ETest#testGenerateHashesIsHermeticAcrossWorkspacePaths` +runs two hashes in one JVM on one machine; `serve_harness.py` drives a single instance. This +harness is the first thing that runs several instances and compares what they each *wrote*. + +Why the S3 tier is the observation point +---------------------------------------- +`HashService.generate` writes the full `{label: hash}` map for a revision to the cache under +`.` before returning. Intercepting those writes gives us, for one commit, +N independently-computed hash maps -- so a failure names the exact targets that diverged rather +than just reporting "the impacted set differed". `mock_s3.py` stands in for the bucket and records +every request; it needs no container and no third-party dependency. + +Each instance always writes under its own `--s3Prefix` (`i0/`, `i1/`, ...) so writes stay +attributable. `--s3Prefix` is deliberately *not* part of `ServeCommand.computeConfigFingerprint()`, +so the instances still agree on the cache key. The mock's read mode then decides what they can see: + + * isolated -- everyone generates independently. This is the poisoning detector. + * shared -- the real fleet topology; followers must hit the entry the first instance published. + +What is deliberately skewed between instances +--------------------------------------------- +Anything that must *not* change a hash, and that is not part of the config fingerprint (skewing a +fingerprinted flag would give the instances different cache keys and there would be nothing to +compare): + + * clone paths of different lengths -- Bazel's output base is `_bazel_/`, so this genuinely gives each instance a different output base and its own Bazel server. + That is the lever for the highest-risk leak: `RuleHasher.ruleBzlSeed` filters macro + instantiation-stack frames only by an `external/` / `@` / `../` prefix, so an absolute `.bzl` + frame would be hashed verbatim -- output base and all. + * `HOME` / `TMPDIR` / `USER` (opt-in, `--env-skew`). + * the Bazel binary (opt-in, `--bazel-alt`) -- the Bazel version is absent from the config + fingerprint, so two versions share a cache key by construction. + +Usage: + tools/serve_consistency.py # build, then run the hermetic matrix + tools/serve_consistency.py --skip-build -v + tools/serve_consistency.py --only baseline + tools/serve_consistency.py --instances 5 + tools/serve_consistency.py --env-skew # also skew HOME/TMPDIR/USER + tools/serve_consistency.py --bazel-alt ~/bin/bazel-8 --only version-skew + tools/serve_consistency.py --real-repo-url --real-from --real-to + +Exit code is non-zero if any gating check fails. Requires git (with `git daemon`) and a bazel +binary. Pure Python stdlib, no third-party deps. + +Run it directly (`python3 tools/serve_consistency.py`), never via `bazel run` -- it shells out to +`bazel build //cli:bazel-diff` and nested bazel deadlocks on the output-base lock. See tools/BUILD. +""" + +from __future__ import annotations + +import argparse +import concurrent.futures +import contextlib +import hashlib +import json +import os +import shutil +import sys +import tempfile +import time +import traceback +from dataclasses import dataclass, field +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +import mock_s3 # noqa: E402 +import serve_harness as base # noqa: E402 + +BUCKET = "bazel-diff-consistency" +S3_REGION = "us-east-1" + +# Flag-shaped option values argparse would otherwise reject; see fold_flag_values. +FLAG_VALUED_OPTS = ("--real-clone-args", "--serve-arg") + + +# ---------------------------------------------------------------------------------------------- +# Environment for a serve instance +# ---------------------------------------------------------------------------------------------- + + +def aws_env() -> dict: + """The AWS wiring every instance needs, plus the guards that make a misconfiguration loud. + + Credentials are mandatory, not cosmetic: the SDK's default provider chain throws + `SdkClientException` when it cannot resolve any, and `S3HashCacheStorage` swallows every + `SdkException` into a silent cache miss -- so without these the harness would run green having + exercised no S3 at all. `AWS_REGION` is even more load-bearing: the region is resolved eagerly + inside `createS3Storage()`, which has no try/catch, so an unresolvable region kills the process + at startup and every instance reports `down`. + + The `/dev/null` config paths and the cleared `AWS_PROFILE` mean a stray `~/.aws/config` on a + developer machine can never redirect this harness at a real bucket. `AWS_MAX_ATTEMPTS=1` + prevents SDK retries from double-counting writes if the mock ever answers badly. + """ + return { + "AWS_ACCESS_KEY_ID": "mock-access-key", + "AWS_SECRET_ACCESS_KEY": "mock-secret-key", + "AWS_REGION": S3_REGION, + "AWS_DEFAULT_REGION": S3_REGION, + "AWS_EC2_METADATA_DISABLED": "true", + "AWS_MAX_ATTEMPTS": "1", + # Shrinks (but does not eliminate) the aws-chunked PutObject path; mock_s3 decodes it + # either way. + "AWS_REQUEST_CHECKSUM_CALCULATION": "when_required", + "AWS_RESPONSE_CHECKSUM_VALIDATION": "when_required", + "AWS_SHARED_CREDENTIALS_FILE": os.devnull, + "AWS_CONFIG_FILE": os.devnull, + "AWS_PROFILE": None, # None removes the variable from the child's environment + "AWS_SESSION_TOKEN": None, + } + + +def skewed_clone_names(n: int, *, skew: bool = True) -> list: + """Directory names of deliberately different lengths, one per instance. + + Bazel derives its output base from an md5 of the absolute workspace path, so different-length + names put each instance on its own output base (and its own Bazel server) without touching any + flag that feeds the config fingerprint. With [skew] off the names are uniform length, which is + the control condition. + """ + if not skew: + return [f"ws-i{i}-pad" for i in range(n)] + return [f"ws-i{i}-" + "p" * (3 + 11 * i) for i in range(n)] + + +def output_base_of(workspace: Path) -> str: + """Bazel's output-base directory name for [workspace]: md5 of the absolute path. + + Reported as an INFO row so "were these actually different-looking hosts?" is answerable from + the run output alone. + """ + return hashlib.md5(str(workspace).encode("utf-8")).hexdigest() + + +# ---------------------------------------------------------------------------------------------- +# One serve instance in the fleet +# ---------------------------------------------------------------------------------------------- + + +@dataclass +class Instance: + idx: int + iid: str # "i0", "i1", ... -- also the S3 prefix stem + clone: Path + cache: Path + home: Path | None = None + tmpdir: Path | None = None + bazel: str | None = None # None => the harness-wide --bazel + seed_file: Path | None = None # --seed-filepaths, used by the negative control + serve: object | None = None + health: str = "unstarted" + + @property + def prefix(self) -> str: + return f"{self.iid}/" + + def env(self, *, env_skew: bool) -> dict: + out = aws_env() + if env_skew: + out["HOME"] = str(self.home) + out["TMPDIR"] = str(self.tmpdir) + out["USER"] = f"bd-consistency-{self.iid}" + out["LOGNAME"] = f"bd-consistency-{self.iid}" + return out + + def s3_flags(self, endpoint: str) -> list: + return [ + "--s3Bucket", BUCKET, + "--s3Prefix", self.prefix, + "--s3Region", S3_REGION, + "--s3Endpoint", endpoint, + "--s3ForcePathStyle", + ] + + def extra_args(self, endpoint: str, serve_args=()) -> list: + args = self.s3_flags(endpoint) + if self.seed_file is not None: + args += ["--seed-filepaths", str(self.seed_file)] + # Applied identically to every instance. Uniformity is what keeps this safe: several of + # these flags feed computeConfigFingerprint(), so passing them to only some instances would + # split the fleet across cache keys and leave nothing to compare. + args += [str(a) for a in serve_args] + return args + + +# ---------------------------------------------------------------------------------------------- +# Payload comparison -- the load-bearing logic (pure; unit-tested) +# ---------------------------------------------------------------------------------------------- + + +def decode_payload(body: bytes): + """Parses a cached hash entry into (hashes, module_graph_json, dep_edges). + + `HashService.serialize` emits one of two shapes: a bare `{label: hash}` object when there is no + metadata, or `{"hashes": {...}, "metadata": {"moduleGraphJson": ..., "depEdges": {...}}}` when + there is. Labels always start with `//` or `@`, so the `"hashes"` key cannot be ambiguous. + """ + obj = json.loads(body.decode("utf-8")) + if isinstance(obj, dict) and isinstance(obj.get("hashes"), dict): + meta = obj.get("metadata") or {} + return obj["hashes"], meta.get("moduleGraphJson"), meta.get("depEdges") or {} + return obj, None, {} + + +def _generated(label: str) -> bool: + """Generated-file labels, whose presence can fluctuate with Bazel's incremental state.""" + return label.endswith(".out") + + +@dataclass +class Divergence: + """What differs between the payloads several instances wrote for one cache key.""" + + logical_key: str + instances: list = field(default_factory=list) + byte_identical: bool = True + key_order_only: bool = False + # instance -> labels it is missing that at least one other instance has + label_only_in: dict = field(default_factory=dict) + # (label, {instance: hash}) for labels present everywhere with differing values + hash_mismatches: list = field(default_factory=list) + total_hash_mismatches: int = 0 + module_graph_differs: bool = False + dep_edges_differ: bool = False + parse_errors: dict = field(default_factory=dict) + + @property + def stable_label_sets_differ(self) -> bool: + return any( + any(not _generated(label) for label in labels) + for labels in self.label_only_in.values() + ) + + @property + def stable_hash_mismatches(self) -> list: + return [(label, vals) for label, vals in self.hash_mismatches if not _generated(label)] + + def as_json(self) -> dict: + return { + "logical_key": self.logical_key, + "instances": self.instances, + "byte_identical": self.byte_identical, + "key_order_only": self.key_order_only, + "label_only_in": {k: sorted(v) for k, v in self.label_only_in.items()}, + "hash_mismatches": [ + {"label": label, "hashes": vals} for label, vals in self.hash_mismatches + ], + "total_hash_mismatches": self.total_hash_mismatches, + "module_graph_differs": self.module_graph_differs, + "dep_edges_differ": self.dep_edges_differ, + "parse_errors": self.parse_errors, + } + + +def compare_payloads(bodies: dict, *, max_labels: int = 20) -> Divergence: + """Compares the cache payloads [bodies] (instance id -> bytes) written for one key. + + Compares *semantically*, not byte-wise. `HashService.serialize` gsons a `HashMap` produced by + `Collectors.toMap` over a parallel stream, so JSON key order depends on how the ForkJoinPool + happened to split the work -- i.e. on the host's CPU count. Two hosts can legitimately emit the + same hashes in a different order. Byte-identity is still recorded (`byte_identical` / + `key_order_only`) because "identical bytes" is a stronger and useful signal, just not the gate. + """ + div = Divergence(logical_key="", instances=sorted(bodies)) + if len(bodies) < 2: + return div + + raw = list(bodies.values()) + div.byte_identical = all(b == raw[0] for b in raw) + + decoded: dict = {} + for iid, body in bodies.items(): + try: + decoded[iid] = decode_payload(body) + except (ValueError, TypeError, UnicodeDecodeError) as exc: + div.parse_errors[iid] = f"{type(exc).__name__}: {exc}" + if len(decoded) < 2: + return div + + hashes = {iid: d[0] for iid, d in decoded.items()} + graphs = {iid: d[1] for iid, d in decoded.items()} + edges = {iid: d[2] for iid, d in decoded.items()} + + all_labels = set() + for h in hashes.values(): + all_labels |= set(h) + for iid, h in hashes.items(): + missing = all_labels - set(h) + if missing: + div.label_only_in[iid] = sorted(missing) + + shared_labels = set(all_labels) + for h in hashes.values(): + shared_labels &= set(h) + mismatches = [] + for label in sorted(shared_labels): + values = {iid: h[label] for iid, h in hashes.items()} + if len(set(values.values())) > 1: + mismatches.append((label, values)) + div.total_hash_mismatches = len(mismatches) + # Report the stable (non-generated) labels first: those are the ones that gate. + ordered = [m for m in mismatches if not _generated(m[0])] + ordered += [m for m in mismatches if _generated(m[0])] + div.hash_mismatches = ordered[:max_labels] + + div.module_graph_differs = _module_graphs_differ(list(graphs.values())) + normalized_edges = [{k: sorted(v) for k, v in e.items()} for e in edges.values()] + div.dep_edges_differ = any(e != normalized_edges[0] for e in normalized_edges[1:]) + + div.key_order_only = ( + not div.byte_identical + and not div.label_only_in + and not mismatches + and not div.module_graph_differs + and not div.dep_edges_differ + ) + return div + + +def _module_graphs_differ(graphs: list) -> bool: + """Compares `moduleGraphJson` blobs as parsed JSON, falling back to string equality. + + The blob is `bazel mod graph --output=json` output captured verbatim, so comparing it as a + string would flag key-ordering noise as a difference. + """ + if not graphs: + return False + parsed = [] + for g in graphs: + if g is None: + parsed.append(None) + continue + try: + parsed.append(json.loads(g)) + except (ValueError, TypeError): + parsed.append(g) + return any(p != parsed[0] for p in parsed[1:]) + + +def divergence_verdict(div: Divergence) -> tuple: + """Maps a Divergence onto a (status, detail) row. + + Rule and source-file differences gate. Generated-file (`*.out`) label-set differences are an + XFAIL: their membership fluctuates with Bazel's incremental analysis state across the service's + sequential checkouts, which is why `serve_harness._stable` drops them from impacted-target + comparisons too. A payload that differs only in JSON key order passes with a note. + """ + if div.parse_errors: + return base.FAIL, "unparseable payload: " + json.dumps(div.parse_errors) + if div.stable_hash_mismatches or div.stable_label_sets_differ or div.module_graph_differs: + return base.FAIL, format_divergence(div) + if div.dep_edges_differ: + return base.FAIL, "depEdges differ between instances" + format_divergence(div) + if div.label_only_in or div.hash_mismatches: + return base.XFAIL, "generated-file (*.out) differences only: " + format_divergence(div) + if div.byte_identical: + return base.PASS, "byte-identical payloads" + if div.key_order_only: + return base.PASS, "identical hashes (JSON key order differs -- expected)" + return base.PASS, "identical hashes" + + +def summarize_divergence(div: Divergence) -> str: + """One-line shape of a divergence, without the per-label hash dump.""" + parts = [] + if div.module_graph_differs: + parts.append("moduleGraphJson differs") + if div.dep_edges_differ: + parts.append("depEdges differ") + if div.label_only_in: + parts.append( + "label sets differ: " + + ", ".join( + f"{iid} missing {len(labels)}" for iid, labels in sorted(div.label_only_in.items()) + ) + ) + if div.total_hash_mismatches: + example = div.hash_mismatches[0][0] if div.hash_mismatches else "" + parts.append(f"{div.total_hash_mismatches} label(s) hash differently (e.g. {example})") + return "; ".join(parts) if parts else "no differences" + + +def format_divergence(div: Divergence, limit: int = 8) -> str: + """The full report: the one-line shape plus the first [limit] diverging labels, one per line. + + This is the payload of a failing run -- the labels named here are where a host-specific input + entered the hash, so they are the starting point for tracing the leak. + """ + out = summarize_divergence(div) + for label, values in div.hash_mismatches[:limit]: + rendered = " ".join(f"{iid}={h}" for iid, h in sorted(values.items())) + out += f"\n {label}: {rendered}" + if div.total_hash_mismatches > limit: + out += f"\n ... {div.total_hash_mismatches - limit} more" + return out + + +# ---------------------------------------------------------------------------------------------- +# Fleet lifecycle +# ---------------------------------------------------------------------------------------------- + + +@dataclass +class Ctx: + args: argparse.Namespace + root: Path + rep: base.Report + store: mock_s3.MockS3Store + server: mock_s3.MockS3Server + remote: object = None + divergences: dict = field(default_factory=dict) # case -> [Divergence] + output_bases: dict = field(default_factory=dict) + clones: list = field(default_factory=list) + + +def make_instances(ctx: Ctx, case: str, n: int, *, clone_args=()) -> list: + """Clones the remote n times into differently-named directories and registers the prefixes.""" + names = skewed_clone_names(n, skew=not ctx.args.no_path_skew) + instances = [] + for idx, name in enumerate(names): + iid = f"i{idx}" + clone = ctx.root / case / name + clone.parent.mkdir(parents=True, exist_ok=True) + ctx.remote.clone(clone, *clone_args) + inst = Instance( + idx=idx, + iid=iid, + clone=clone, + cache=ctx.root / case / f"cache-{iid}", + home=ctx.root / case / f"home-{iid}", + tmpdir=ctx.root / case / f"tmp-{iid}", + ) + inst.cache.mkdir(parents=True, exist_ok=True) + inst.home.mkdir(parents=True, exist_ok=True) + inst.tmpdir.mkdir(parents=True, exist_ok=True) + ctx.store.register(iid, inst.prefix) + ctx.output_bases[f"{case}/{iid}"] = output_base_of(clone) + ctx.clones.append(clone) + instances.append(inst) + return instances + + +@contextlib.contextmanager +def fleet(ctx: Ctx, instances: list, *, track_deps: bool): + """Starts every instance, yields the list, and always tears them down. + + Instances come up one at a time (`base.serve` blocks on the health handshake before yielding), + which keeps N cold Bazel servers from starting simultaneously. With `--no-initial-fetch` and no + warmup revision, readiness is immediate, so this costs almost nothing. + """ + ports = base.reserve_ports(len(instances)) + with contextlib.ExitStack() as stack: + for inst, port in zip(instances, ports): + cm = base.serve( + inst.clone, + inst.cache, + initial_fetch=False, + track_deps=track_deps, + ready_timeout=ctx.args.ready_timeout, + verbose_server=True, + extra_args=inst.extra_args(ctx.server.url, ctx.args.serve_arg), + env=inst.env(env_skew=ctx.args.env_skew), + port=port, + bazel=inst.bazel, + ) + served, health = stack.enter_context(cm) + inst.serve = served + inst.health = health + yield instances + + +def all_ready(ctx: Ctx, case: str, instances: list) -> bool: + ok = True + for inst in instances: + if not ctx.rep.check( + case, + f"{inst.iid} became ready", + inst.health == "ready", + ok_detail=f"port {inst.serve.port}, output_base {output_base_of(inst.clone)[:12]}", + fail_detail=f"health={inst.health}; stderr:\n{inst.serve.stderr_tail(15)}", + ): + ok = False + return ok + + +def report_query_failures(ctx: Ctx, case: str, instances: list, results: dict) -> bool: + """Surfaces the server-side error when a query failed. Returns True if every instance answered. + + Called before `assert_wired`, because a failed query writes nothing to the cache and would + otherwise trip the "no PutObject reached the mock" assertion -- reporting a broken S3 tier when + the real problem was an exception inside the hash generation. The instance's stderr holds the + actual stack trace, so it goes in the failure detail. + """ + failed = {iid: code for iid, (code, _b, _d) in results.items() if code != 200} + if not failed: + return True + by_iid = {inst.iid: inst for inst in instances} + for iid, code in sorted(failed.items()): + inst = by_iid.get(iid) + error_lines = [ + line + for line in (inst.serve.stderr_tail(200).splitlines() if inst else []) + if "Error" in line or "Exception" in line or "ERROR:" in line + ] + ctx.rep.add( + case, + f"{iid} query failed", + base.FAIL, + f"HTTP {code}; server said: " + (" | ".join(error_lines[:4]) or "(no error in stderr)"), + ) + return False + + +def assert_wired(ctx: Ctx, case: str, instances: list) -> bool: + """Proves the S3 tier is actually in the loop before any consistency claim is believed. + + `S3HashCacheStorage` contains every failure -- a read error becomes a miss, a write error a + warning -- so an instance whose S3 client is broken behaves exactly like one whose cache is + simply cold. Without these checks a wholly disconnected fleet would report "no divergence". + """ + ok = True + for inst in instances: + code, body = base.http(inst.serve.port, "/metrics", timeout=30) + remote = None + with contextlib.suppress(ValueError, TypeError, AttributeError): + remote = (json.loads(body).get("cache") or {}).get("remote") + expected = f"s3://{BUCKET}/{inst.prefix}" + ok &= ctx.rep.check( + case, + f"{inst.iid} reports the remote cache tier", + code == 200 and remote == expected, + ok_detail=remote or "", + fail_detail=f"code={code} cache.remote={remote!r} expected={expected!r}", + ) + ok &= ctx.rep.check( + case, + f"{inst.iid} wrote to the mock S3", + bool(ctx.store.writes(instance=inst.iid)), + ok_detail=f"{len(ctx.store.writes(instance=inst.iid))} object(s)", + fail_detail="no PutObject reached the mock -- the S3 tier is not in the loop", + ) + ok &= ctx.rep.check( + case, + f"{inst.iid} read from the mock S3", + bool(ctx.store.reads(instance=inst.iid)), + ok_detail=f"{len(ctx.store.reads(instance=inst.iid))} request(s)", + fail_detail="no GetObject/HeadObject reached the mock", + ) + swallowed = [ + line + for line in inst.serve.stderr_tail(400).splitlines() + if "S3 cache" in line and "failed" in line + ] + ok &= ctx.rep.check( + case, + f"{inst.iid} logged no swallowed S3 errors", + not swallowed, + fail_detail="; ".join(swallowed[:3]), + ) + ok &= ctx.rep.check( + case, + "mock S3 saw no unexpected requests", + not ctx.store.bad_requests(), + fail_detail=str(ctx.store.bad_requests()[:3]), + ) + return ok + + +def chunked_transport_check(ctx: Ctx, case: str) -> None: + """Records whether PutObject arrived aws-chunked. + + Informational, but worth surfacing: if a future SDK bump changes the framing and the decoder + silently starts recording raw bytes, every payload would differ and this row is the first place + to look. + """ + writes = ctx.store.writes() + if not writes: + return + chunked = sum(1 for w in writes if w.chunked) + ctx.rep.add( + case, + "PutObject transfer framing", + base.INFO, + f"{chunked}/{len(writes)} writes arrived aws-chunked", + ) + + +def compare_all_keys(ctx: Ctx, case: str, instances: list, *, expect_divergence: bool = False): + """Diffs the payloads every instance wrote, per cache key. The core assertion.""" + keys = ctx.store.logical_keys() + ctx.rep.check( + case, + "instances agree on the cache key(s)", + bool(keys), + ok_detail=f"{len(keys)} key(s): {', '.join(k[:20] for k in keys)}", + fail_detail="no cache keys were written at all", + ) + + diverged_any = False + for key in keys: + bodies = ctx.store.bodies_for(key) + if len(bodies) < 2: + ctx.rep.add( + case, + f"key {key[:16]}... written by only {len(bodies)} instance(s)", + base.INFO, + # In isolated mode this means an instance failed to publish; in shared mode it is + # the expected outcome of a cache hit. + f"wrote: {', '.join(sorted(bodies)) or 'nobody'}", + ) + continue + div = compare_payloads(bodies, max_labels=ctx.args.max_diff_labels) + div.logical_key = key + ctx.divergences.setdefault(case, []).append(div) + status, detail = divergence_verdict(div) + if status == base.FAIL: + diverged_any = True + if expect_divergence: + # Polarity is inverted: the negative control passes when divergence IS detected. The + # detail stays short here -- a detected-as-expected difference is not a finding. + ctx.rep.check( + case, + f"divergence detected for {key[:16]}...", + status == base.FAIL, + ok_detail=summarize_divergence(div), + fail_detail="no divergence detected -- the comparator is not actually comparing", + ) + else: + ctx.rep.add(case, f"payloads agree for {key[:16]}...", status, detail) + + if not expect_divergence: + sizes = { + iid: len(decode_payload(body)[0]) + for key in keys + for iid, body in ctx.store.bodies_for(key).items() + } + if sizes: + ctx.rep.add( + case, + "target counts per instance", + base.INFO, + ", ".join(f"{iid}={n}" for iid, n in sorted(sizes.items())), + ) + return diverged_any + + +def compare_impacted(ctx: Ctx, case: str, results: dict) -> None: + """The user-visible symptom: do the instances return the same impacted targets?""" + sets = {iid: base._stable(data) for iid, (code, _body, data) in results.items() if data} + codes = {iid: code for iid, (code, _b, _d) in results.items()} + ctx.rep.check( + case, + "every instance answered 200", + all(c == 200 for c in codes.values()), + ok_detail=f"{len(codes)} responses", + fail_detail=str(codes), + ) + if len(sets) < 2: + return + first_iid = sorted(sets)[0] + reference = sets[first_iid] + mismatched = {iid: s for iid, s in sets.items() if s != reference} + detail = f"{len(reference)} stable targets" + if mismatched: + lines = [] + for iid, s in sorted(mismatched.items()): + lines.append( + f"{iid}: +{sorted(s - reference)[:5]} -{sorted(reference - s)[:5]}" + ) + detail = f"vs {first_iid}: " + "; ".join(lines) + ctx.rep.check( + case, + "impacted targets identical across instances", + not mismatched, + ok_detail=detail, + fail_detail=detail, + ) + + +def impacted_all( + instances: list, + frm: str, + to: str, + *, + concurrent_requests: bool = False, + timeout: float = 300.0, +) -> dict: + """POSTs the same query to every instance (sequentially by default) with profiling on. + + Sequential is the default so N cold Bazel servers do not analyze at once; only the + `concurrent` case, which is specifically about the race, fires them together. [timeout] needs + raising for a real repository -- a cold output base fetching external repos can take far longer + than the synthetic workspace's few seconds. + """ + if not concurrent_requests: + return { + inst.iid: base._impacted_post(inst.serve, frm, to, profile=True, timeout=timeout) + for inst in instances + } + results: dict = {} + with concurrent.futures.ThreadPoolExecutor(max_workers=len(instances)) as pool: + futures = { + pool.submit( + base._impacted_post, inst.serve, frm, to, profile=True, timeout=timeout + ): inst.iid + for inst in instances + } + for fut in concurrent.futures.as_completed(futures): + results[futures[fut]] = fut.result() + return results + + +def retrievals(data) -> list: + """The `profile.hashRetrievals` list from an /impacted_targets response, or [].""" + if not isinstance(data, dict): + return [] + profile = data.get("profile") or {} + got = profile.get("hashRetrievals") + return got if isinstance(got, list) else [] + + +# ---------------------------------------------------------------------------------------------- +# Cases +# ---------------------------------------------------------------------------------------------- + + +def case_baseline(ctx: Ctx) -> None: + """N instances, isolated reads: everyone hashes the same commits independently.""" + case = "baseline" + base.log(f"\n{base.C.BOLD}[{case}] {ctx.args.instances} instances, isolated cache{base.C.RESET}") + ctx.store.reset() + ctx.store.set_mode(mock_s3.ISOLATED) + instances = make_instances(ctx, case, ctx.args.instances) + frm, to = ctx.remote.shas["C1"], ctx.remote.shas["C2"] + + with fleet(ctx, instances, track_deps=ctx.args.track_deps): + if not all_ready(ctx, case, instances): + return + results = impacted_all(instances, frm, to, timeout=ctx.args.request_timeout) + if not report_query_failures(ctx, case, instances, results): + return + assert_wired(ctx, case, instances) + chunked_transport_check(ctx, case) + compare_all_keys(ctx, case, instances) + compare_impacted(ctx, case, results) + ctx.rep.add( + case, + "output bases", + base.INFO, + ", ".join(f"{i.iid}={output_base_of(i.clone)[:10]}" for i in instances), + ) + _shutdown(ctx, instances) + + +def case_negative_control(ctx: Ctx) -> None: + """Proves the detector is live by making one instance's hashes genuinely differ. + + `--seed-filepaths` is hashed into every target + (`BuildGraphHasher.createSeedForFilepaths`) but is absent from + `ServeCommand.computeConfigFingerprint()`, so giving one instance a different seed file changes + all of its hashes while leaving the cache key untouched -- exactly the shape of a real + poisoning bug, and a genuine product observation in its own right (see the INFO row). + """ + case = "negative-control" + base.log(f"\n{base.C.BOLD}[{case}] one instance seeded differently{base.C.RESET}") + ctx.store.reset() + ctx.store.set_mode(mock_s3.ISOLATED) + instances = make_instances(ctx, case, max(2, ctx.args.instances)) + + # The seed content lives outside every clone, so `git checkout --force` cannot disturb it. + seeds = ctx.root / case / "seeds" + seeds.mkdir(parents=True, exist_ok=True) + for inst in instances: + content = seeds / f"seed-{inst.iid}.txt" + content.write_text("shared seed\n" if inst.idx == 0 else f"divergent seed {inst.iid}\n") + listing = seeds / f"seedlist-{inst.iid}.txt" + listing.write_text(f"{content}\n") + inst.seed_file = listing + # i0 and i1 differ; that is enough for the comparator to have something to find. + (seeds / "seed-i1.txt").write_text("a genuinely different seed\n") + + frm, to = ctx.remote.shas["C1"], ctx.remote.shas["C2"] + with fleet(ctx, instances, track_deps=ctx.args.track_deps): + if not all_ready(ctx, case, instances): + return + impacted_all(instances, frm, to, timeout=ctx.args.request_timeout) + compare_all_keys(ctx, case, instances, expect_divergence=True) + ctx.rep.add( + case, + "seedFilepaths is absent from the config fingerprint", + base.INFO, + "instances with different --seed-filepaths share a cache key while producing " + "different hashes (ServeCommand.computeConfigFingerprint)", + ) + _shutdown(ctx, instances) + + +def case_shared(ctx: Ctx) -> None: + """The real fleet topology: i0 publishes, the followers must be served by the remote tier.""" + case = "shared" + base.log(f"\n{base.C.BOLD}[{case}] shared cache, follower cache-hit propagation{base.C.RESET}") + ctx.store.reset() + ctx.store.set_mode(mock_s3.SHARED) + instances = make_instances(ctx, case, max(2, ctx.args.instances)) + frm, to = ctx.remote.shas["C0"], ctx.remote.shas["C1"] + + leader, followers = instances[0], instances[1:] + + with fleet(ctx, [leader], track_deps=ctx.args.track_deps): + if not all_ready(ctx, case, [leader]): + return + code, _body, data = base._impacted_post(leader.serve, frm, to, profile=True) + ctx.rep.check( + case, + "leader generated the entries (cache miss)", + code == 200 and all(not r.get("cacheHit") for r in retrievals(data)), + ok_detail=f"{len(retrievals(data))} retrieval(s), all misses", + fail_detail=f"code={code} retrievals={retrievals(data)}", + ) + leader_targets = base._stable(data) + published = ctx.store.writes(instance=leader.iid) + if not ctx.rep.check( + case, + "leader published to the shared tier", + bool(published), + ok_detail=f"{len(published)} object(s)", + fail_detail="nothing reached the mock; followers would have nothing to hit", + ): + _shutdown(ctx, [leader]) + return + _shutdown(ctx, [leader]) + + with fleet(ctx, followers, track_deps=ctx.args.track_deps): + if not all_ready(ctx, case, followers): + return + for inst in followers: + code, _body, data = base._impacted_post(inst.serve, frm, to, profile=True) + rets = retrievals(data) + ctx.rep.check( + case, + f"{inst.iid} served entirely from cache", + code == 200 and bool(rets) and all(r.get("cacheHit") for r in rets), + ok_detail=f"{len(rets)} retrieval(s), all hits", + fail_detail=f"code={code} retrievals={rets}", + ) + # A local-disk hit also reports cacheHit=true, so the profile alone proves nothing. + # The mock's read log is what shows the *remote* tier served it, and whose write did. + remote_hits = [ + r + for r in ctx.store.reads(instance=inst.iid, op="GET") + if r.hit and r.served_from_prefix == leader.prefix + ] + ctx.rep.check( + case, + f"{inst.iid} was served by {leader.iid}'s objects over S3", + bool(remote_hits), + ok_detail=f"{len(remote_hits)} remote hit(s)", + fail_detail="no GET resolved to the leader's prefix -- a local-tier hit, not remote", + ) + ctx.rep.check( + case, + f"{inst.iid} did not re-publish on a hit", + not ctx.store.writes(instance=inst.iid), + fail_detail=f"{len(ctx.store.writes(instance=inst.iid))} redundant write(s)", + ) + ctx.rep.check( + case, + f"{inst.iid} impacted targets match the leader", + base._stable(data) == leader_targets, + ok_detail=f"{len(leader_targets)} stable targets", + fail_detail=( + f"+{sorted(base._stable(data) - leader_targets)[:5]} " + f"-{sorted(leader_targets - base._stable(data))[:5]}" + ), + ) + _shutdown(ctx, followers) + + +def case_concurrent(ctx: Ctx) -> None: + """All instances race the same cold revision pair against one shared bucket. + + Cross-process generation is unlocked by design (`HashService.generationLock` is per-JVM), so + several instances generating the same key is expected and benign. Divergent *content* under one + key is the bug. + """ + case = "concurrent" + base.log(f"\n{base.C.BOLD}[{case}] all instances race one cold revision{base.C.RESET}") + ctx.store.reset() + ctx.store.set_mode(mock_s3.SHARED) + instances = make_instances(ctx, case, ctx.args.instances) + frm, to = ctx.remote.shas["C0"], ctx.remote.shas["C2"] + + with fleet(ctx, instances, track_deps=ctx.args.track_deps): + if not all_ready(ctx, case, instances): + return + results = impacted_all( + instances, frm, to, concurrent_requests=True, timeout=ctx.args.request_timeout + ) + compare_impacted(ctx, case, results) + for key in ctx.store.logical_keys(): + writes = ctx.store.writes(logical_key=key) + digests = {w.sha256 for w in writes} + ctx.rep.check( + case, + f"racing writers agree on {key[:16]}...", + len(digests) <= 1, + ok_detail=f"{len(writes)} write(s), 1 distinct payload", + fail_detail=( + f"{len(writes)} write(s) produced {len(digests)} distinct payloads:\n" + + format_divergence( + compare_payloads( + ctx.store.bodies_for(key), max_labels=ctx.args.max_diff_labels + ) + ) + ), + ) + generated = sum( + 1 + for _iid, (_c, _b, data) in results.items() + if any(not r.get("cacheHit") for r in retrievals(data)) + ) + ctx.rep.add( + case, + "race outcome", + base.INFO, + f"{generated}/{len(results)} instances generated the hashes themselves; " + f"{len(results) - generated} were served by the shared tier", + ) + _shutdown(ctx, instances) + + +def case_version_skew(ctx: Ctx) -> None: + """Two instances, identical config, different Bazel binaries. + + The Bazel version is not part of `computeConfigFingerprint()`, yet it changes the query + pipeline materially (`BazelQueryService` gates `streamed_proto`, `--output_file` and + `mod show_repo` on version thresholds). If the payloads diverge, two Bazel versions are sharing + a cache key -- a real product bug, reported as XFAIL rather than a harness failure. + """ + case = "version-skew" + if not ctx.args.bazel_alt: + ctx.rep.add(case, "skipped", base.SKIP, "pass --bazel-alt to run this case") + return + base.log(f"\n{base.C.BOLD}[{case}] two Bazel versions, one cache key{base.C.RESET}") + ctx.store.reset() + ctx.store.set_mode(mock_s3.ISOLATED) + instances = make_instances(ctx, case, 2) + instances[1].bazel = ctx.args.bazel_alt + frm, to = ctx.remote.shas["C1"], ctx.remote.shas["C2"] + + with fleet(ctx, instances, track_deps=ctx.args.track_deps): + if not all_ready(ctx, case, instances): + return + impacted_all(instances, frm, to, timeout=ctx.args.request_timeout) + keys = ctx.store.logical_keys() + ctx.rep.add( + case, + "both Bazel versions computed the same cache key", + base.INFO if len(keys) else base.FAIL, + f"{len(keys)} distinct key(s) -- the version is absent from the config fingerprint", + ) + for key in keys: + bodies = ctx.store.bodies_for(key) + if len(bodies) < 2: + continue + div = compare_payloads(bodies, max_labels=ctx.args.max_diff_labels) + div.logical_key = key + ctx.divergences.setdefault(case, []).append(div) + status, detail = divergence_verdict(div) + if status == base.FAIL: + ctx.rep.add( + case, + f"Bazel versions disagree on {key[:16]}...", + base.XFAIL, + "two Bazel versions share a cache key but hash differently: " + detail, + ) + else: + ctx.rep.add(case, f"Bazel versions agree on {key[:16]}...", base.PASS, detail) + _shutdown(ctx, instances) + + +def case_real(ctx: Ctx) -> None: + """The `baseline` comparison against a real repository at real revisions.""" + case = "real" + args = ctx.args + source = args.real_repo_url or args.real_repo_path + if not source or not args.real_from or not args.real_to: + ctx.rep.add( + case, + "skipped", + base.SKIP, + "pass --real-repo-url/--real-repo-path plus --real-from and --real-to", + ) + return + base.log(f"\n{base.C.BOLD}[{case}] {source} {args.real_from[:12]}..{args.real_to[:12]}{base.C.RESET}") + ctx.store.reset() + ctx.store.set_mode(mock_s3.ISOLATED) + + clone_args = args.real_clone_args.split() if args.real_clone_args else [] + names = skewed_clone_names(args.instances, skew=not args.no_path_skew) + instances = [] + for idx, name in enumerate(names): + iid = f"i{idx}" + clone = ctx.root / case / name + clone.parent.mkdir(parents=True, exist_ok=True) + base.run(["git", "clone", "-q", *clone_args, str(source), str(clone)]) + base.git(clone, "config", "gc.auto", "0") + inst = Instance( + idx=idx, + iid=iid, + clone=clone, + cache=ctx.root / case / f"cache-{iid}", + home=ctx.root / case / f"home-{iid}", + tmpdir=ctx.root / case / f"tmp-{iid}", + ) + for d in (inst.cache, inst.home, inst.tmpdir): + d.mkdir(parents=True, exist_ok=True) + ctx.store.register(iid, inst.prefix) + ctx.output_bases[f"{case}/{iid}"] = output_base_of(clone) + ctx.clones.append(clone) + instances.append(inst) + + with fleet(ctx, instances, track_deps=ctx.args.track_deps): + if not all_ready(ctx, case, instances): + return + results = impacted_all( + instances, args.real_from, args.real_to, timeout=args.request_timeout + ) + if not report_query_failures(ctx, case, instances, results): + return + assert_wired(ctx, case, instances) + chunked_transport_check(ctx, case) + compare_all_keys(ctx, case, instances) + compare_impacted(ctx, case, results) + _shutdown(ctx, instances) + + +def _shutdown(ctx: Ctx, instances: list) -> None: + """Stops each instance's Bazel server. Must happen at the case boundary, not just at exit: + every instance holds its own output base, so leaving N servers resident across cases is what + turns a 3-instance run into a memory problem on a CI runner.""" + for inst in instances: + base._bazel_shutdown(inst.clone) + + +CASES = [ + ("baseline", case_baseline), + ("negative-control", case_negative_control), + ("shared", case_shared), + ("concurrent", case_concurrent), + ("version-skew", case_version_skew), + ("real", case_real), +] + + +# ---------------------------------------------------------------------------------------------- +# Output: metrics JSON + markdown summary +# ---------------------------------------------------------------------------------------------- + + +def build_metrics_json(ctx: Ctx, wall_seconds: float, failures: int) -> dict: + return { + "meta": { + "harness": "serve_consistency", + "instances": ctx.args.instances, + "bazel": base.BAZEL, + "bazel_alt": ctx.args.bazel_alt, + "path_skew": not ctx.args.no_path_skew, + "env_skew": ctx.args.env_skew, + "track_deps": ctx.args.track_deps, + "serve_args": list(ctx.args.serve_arg), + "wall_seconds": round(wall_seconds, 1), + "python": sys.version.split()[0], + }, + "s3": { + # The store is reset at each case boundary, so `final_case` describes only the last + # case that ran; `run_total` spans every case. + "run_total": ctx.store.total_counts(), + "final_case": ctx.store.counts(), + "final_case_per_instance": ctx.store.counts_by_instance(), + "bad_requests": ctx.store.bad_requests(), + }, + "output_bases": ctx.output_bases, + "divergences": { + case: [d.as_json() for d in divs] for case, divs in ctx.divergences.items() + }, + "checks": [ + {"case": c, "name": n, "status": st, "detail": d} for c, n, st, d in ctx.rep.rows + ], + "failures": failures, + } + + +def build_summary_md(data: dict) -> str: + meta = data["meta"] + verdict = ( + "✅ hashes are consistent across instances" + if data["failures"] == 0 + else f"❌ {data['failures']} check(s) failed" + ) + lines = [ + "# `bazel-diff serve` cross-instance hash consistency", + "", + f"**{verdict}** — {meta['instances']} instances, wall time {meta['wall_seconds']}s, " + f"path skew {'on' if meta['path_skew'] else 'off'}, " + f"env skew {'on' if meta['env_skew'] else 'off'}", + "", + "## Checks", + "", + "| case | check | status | detail |", + "|---|---|---|---|", + ] + for row in data["checks"]: + detail = row["detail"].replace("\n", " ").replace("|", "\\|") + lines.append( + f"| {row['case']} | {row['name']} | {row['status']} | {detail[:300]} |" + ) + + diverged = { + case: [d for d in divs if d["total_hash_mismatches"] or d["label_only_in"]] + for case, divs in data.get("divergences", {}).items() + } + diverged = {k: v for k, v in diverged.items() if v} + if diverged: + lines += ["", "## Diverging targets", ""] + for case, divs in diverged.items(): + for div in divs: + lines += [ + f"### `{case}` — key `{div['logical_key']}`", + "", + f"{div['total_hash_mismatches']} label(s) hash differently; " + f"moduleGraph differs: {div['module_graph_differs']}; " + f"depEdges differ: {div['dep_edges_differ']}", + "", + "```", + ] + for entry in div["hash_mismatches"]: + rendered = " ".join( + f"{iid}={h}" for iid, h in sorted(entry["hashes"].items()) + ) + lines.append(f"{entry['label']}: {rendered}") + lines += ["```", ""] + + s3 = data["s3"]["run_total"] + lines += [ + "", + "## Mock S3 traffic (whole run)", + "", + f"{s3['puts']} PutObject, {s3['gets']} GetObject, {s3['heads']} HeadObject, " + f"{s3['hits']} hit(s), {s3['bad_requests']} unexpected request(s).", + "", + "## Bazel output bases", + "", + "| instance | output base |", + "|---|---|", + ] + for name, digest in sorted(data["output_bases"].items()): + lines.append(f"| {name} | `{digest}` |") + return "\n".join(lines) + "\n" + + +# ---------------------------------------------------------------------------------------------- +# CLI +# ---------------------------------------------------------------------------------------------- + + +def fold_flag_values(argv, opts=FLAG_VALUED_OPTS) -> list: + """Rewrites `--opt ` into `--opt=`. + + argparse rejects an option value that starts with `-` unless it is spelled with `=`, which is + exactly how `--real-clone-args --depth=50` fails. `serve_stress.py` carries the same helper for + the same reason; it broke the serve-stress cron twice before it was added. + """ + out: list = [] + i = 0 + while i < len(argv): + token = argv[i] + if token in opts and i + 1 < len(argv) and argv[i + 1].startswith("-"): + out.append(f"{token}={argv[i + 1]}") + i += 2 + continue + out.append(token) + i += 1 + return out + + +def build_parser() -> argparse.ArgumentParser: + ap = argparse.ArgumentParser( + description="Check that several `bazel-diff serve` instances hash a commit identically." + ) + ap.add_argument("--instances", type=int, default=3, help="fleet size (default 3)") + ap.add_argument("--only", default="", help="run only cases whose name contains this substring") + ap.add_argument("--skip-build", action="store_true", help="reuse existing bazel-bin launcher") + ap.add_argument("--keep-artifacts", action="store_true", help="do not delete the temp workdir") + ap.add_argument("--bazel", default="bazel", help="bazel binary (e.g. a bazelisk path on CI)") + ap.add_argument( + "--bazel-alt", default="", help="second bazel binary; enables the version-skew case") + ap.add_argument( + "--track-deps", action="store_true", help="run instances with --trackDeps (compares depEdges)") + ap.add_argument( + "--no-path-skew", + action="store_true", + help="give every clone the same-length path (control condition)", + ) + ap.add_argument( + "--env-skew", + action="store_true", + help="also skew HOME/TMPDIR/USER between instances (note: a skewed USER makes Bazel " + "create additional output user roots outside the temp workdir, which the harness " + "shuts down but does not delete)", + ) + ap.add_argument( + "--max-diff-labels", type=int, default=20, help="labels reported per divergence (default 20)") + ap.add_argument( + "--ready-timeout", type=float, default=90.0, help="seconds to wait for /health (default 90)") + ap.add_argument( + "--request-timeout", + type=float, + default=300.0, + help="seconds to wait for an /impacted_targets response (default 300; raise it for a real " + "repository, where a cold output base must fetch external repos before it can query)", + ) + ap.add_argument( + "--serve-arg", + action="append", + default=[], + metavar="ARG", + help="extra `serve` flag applied identically to every instance; repeatable. Use the `=` " + "spelling for flag-shaped values, e.g. " + "--serve-arg=--fineGrainedHashExternalRepos=@rules_kotlin", + ) + ap.add_argument("--metrics-out", default="", help="write the run's metrics JSON here") + ap.add_argument("--summary-out", default="", help="write a markdown summary here") + ap.add_argument("-v", "--verbose", action="store_true", help="stream serve/git output") + + real = ap.add_argument_group("real repository mode") + real.add_argument("--real-repo-url", default="", help="clone URL for the `real` case") + real.add_argument("--real-repo-path", default="", help="local repo path for the `real` case") + real.add_argument("--real-from", default="", help="base revision") + real.add_argument("--real-to", default="", help="head revision") + real.add_argument("--real-clone-args", default="", help="extra `git clone` args, space separated") + return ap + + +def parse_args(argv=None) -> argparse.Namespace: + argv = list(sys.argv[1:] if argv is None else argv) + return build_parser().parse_args(fold_flag_values(argv)) + + +def main() -> int: + args = parse_args() + base.VERBOSE = args.verbose + base.BAZEL = args.bazel + if args.instances < 2: + base.log(f"{base.C.RED}--instances must be at least 2{base.C.RESET}") + return 2 + + if not args.skip_build: + base.log(f"{base.C.BOLD}Building //cli:bazel-diff ...{base.C.RESET}") + base.run([base.BAZEL, "build", "//cli:bazel-diff"], cwd=base.REPO_ROOT) + if not base.LAUNCHER.exists(): + base.log(f"{base.C.RED}launcher not found at {base.LAUNCHER}; drop --skip-build{base.C.RESET}") + return 2 + + root = Path(tempfile.mkdtemp(prefix="serve-consistency-")) + base.log(f"{base.C.DIM}workdir: {root}{base.C.RESET}") + started = time.time() + rep = base.Report() + store = mock_s3.MockS3Store(BUCKET) + server = mock_s3.MockS3Server(store) + server.start() + ctx = Ctx(args=args, root=root, rep=rep, store=store, server=server) + base.log(f"{base.C.DIM}mock S3: {server.url}/{BUCKET}{base.C.RESET}") + + selected = [(name, fn) for name, fn in CASES if not args.only or args.only in name] + try: + base.log(f"{base.C.BOLD}Setting up git remote (git daemon) ...{base.C.RESET}") + ctx.remote = base.build_remote(root) + # C3 gives the synthetic history a revision beyond the clone point; the cases here use + # only pre-clone revisions, so no instance needs a fetch (`--no-initial-fetch`). + ctx.remote.shas["C3"] = ctx.remote.commit("C3 restore core (A'')", core=base.CORE_A2) + ctx.remote.repack() + base.vlog(f"shas: {ctx.remote.shas}") + + for name, fn in selected: + try: + fn(ctx) + except Exception: + rep.add(name, "case crashed", base.FAIL, traceback.format_exc().splitlines()[-1]) + base.vlog(traceback.format_exc()) + finally: + for clone in ctx.clones: + base._bazel_shutdown(clone) + if ctx.remote: + ctx.remote.stop() + server.stop() + if args.keep_artifacts: + base.log(f"{base.C.DIM}artifacts kept at {root}{base.C.RESET}") + else: + shutil.rmtree(root, ignore_errors=True) + + failures = rep.summary() + data = build_metrics_json(ctx, time.time() - started, failures) + if args.metrics_out: + Path(args.metrics_out).write_text(json.dumps(data, indent=2) + "\n") + base.log(f"{base.C.DIM}metrics written to {args.metrics_out}{base.C.RESET}") + if args.summary_out: + Path(args.summary_out).write_text(build_summary_md(data)) + base.log(f"{base.C.DIM}summary written to {args.summary_out}{base.C.RESET}") + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/serve_consistency_test.py b/tools/serve_consistency_test.py new file mode 100644 index 00000000..e36cc45e --- /dev/null +++ b/tools/serve_consistency_test.py @@ -0,0 +1,360 @@ +#!/usr/bin/env python3 +"""Unit tests for the pure helpers in tools/serve_consistency.py. + +Only the functions that never shell out are exercised here -- importing the module is safe, but +running a case would invoke `bazel build`, and nested bazel deadlocks on the output-base lock (see +the comment block in tools/BUILD). + +`compare_payloads` / `divergence_verdict` are the load-bearing pair: they decide whether a run +reports "hashes are consistent" or names diverging targets, so they are tested against the two +payload shapes `HashService.serialize` emits and against the failure modes that would otherwise +produce a false green -- most importantly the JSON key-order difference that is *expected* between +hosts and must not be reported as divergence. +""" + +import json +import unittest + +import serve_consistency as sc + + +def _payload(hashes, module_graph=None, dep_edges=None, *, wrapped=None): + """Serializes [hashes] the way HashService.serialize does.""" + if wrapped is None: + wrapped = module_graph is not None or dep_edges is not None + if not wrapped: + return json.dumps(hashes).encode() + metadata = {} + if module_graph is not None: + metadata["moduleGraphJson"] = module_graph + if dep_edges is not None: + metadata["depEdges"] = dep_edges + return json.dumps({"hashes": hashes, "metadata": metadata}).encode() + + +HASHES = { + "//:core": "Rule#aaa~bbb", + "//:mid": "Rule#ccc~ddd", + "//:core.txt": "SourceFile#eee~fff", +} + + +class DecodePayloadTest(unittest.TestCase): + def test_bare_hash_map(self): + hashes, graph, edges = sc.decode_payload(_payload(HASHES)) + self.assertEqual(hashes, HASHES) + self.assertIsNone(graph) + self.assertEqual(edges, {}) + + def test_wrapped_with_metadata(self): + body = _payload(HASHES, module_graph='{"root":"x"}', dep_edges={"//:mid": ["//:core"]}) + hashes, graph, edges = sc.decode_payload(body) + self.assertEqual(hashes, HASHES) + self.assertEqual(graph, '{"root":"x"}') + self.assertEqual(edges, {"//:mid": ["//:core"]}) + + +class ComparePayloadsTest(unittest.TestCase): + def test_identical_payloads_pass(self): + div = sc.compare_payloads({"i0": _payload(HASHES), "i1": _payload(HASHES)}) + self.assertTrue(div.byte_identical) + self.assertEqual(div.hash_mismatches, []) + status, _ = sc.divergence_verdict(div) + self.assertEqual(status, sc.base.PASS) + + def test_key_order_difference_is_not_divergence(self): + # HashService.serialize gsons a HashMap built by Collectors.toMap over a *parallel* stream, + # so key order tracks the host's CPU count. Two hosts may legitimately differ here. + reordered = dict(reversed(list(HASHES.items()))) + div = sc.compare_payloads({"i0": _payload(HASHES), "i1": _payload(reordered)}) + self.assertFalse(div.byte_identical) + self.assertTrue(div.key_order_only) + status, detail = sc.divergence_verdict(div) + self.assertEqual(status, sc.base.PASS) + self.assertIn("key order", detail) + + def test_single_label_hash_difference_fails(self): + other = dict(HASHES, **{"//:core": "Rule#zzz~bbb"}) + div = sc.compare_payloads({"i0": _payload(HASHES), "i1": _payload(other)}) + self.assertEqual(div.total_hash_mismatches, 1) + self.assertEqual(div.hash_mismatches[0][0], "//:core") + self.assertEqual( + div.hash_mismatches[0][1], {"i0": "Rule#aaa~bbb", "i1": "Rule#zzz~bbb"} + ) + status, detail = sc.divergence_verdict(div) + self.assertEqual(status, sc.base.FAIL) + self.assertIn("//:core", detail) + + def test_label_set_difference_fails(self): + fewer = {k: v for k, v in HASHES.items() if k != "//:mid"} + div = sc.compare_payloads({"i0": _payload(HASHES), "i1": _payload(fewer)}) + self.assertEqual(div.label_only_in, {"i1": ["//:mid"]}) + self.assertTrue(div.stable_label_sets_differ) + self.assertEqual(sc.divergence_verdict(div)[0], sc.base.FAIL) + + def test_generated_file_only_difference_is_xfail(self): + # `*.out` membership fluctuates with Bazel's incremental analysis state across the + # service's sequential checkouts; serve_harness._stable drops them for the same reason. + with_out = dict(HASHES, **{"//:core.out": "GeneratedFile#aaa~bbb"}) + div = sc.compare_payloads({"i0": _payload(HASHES), "i1": _payload(with_out)}) + self.assertFalse(div.stable_label_sets_differ) + self.assertEqual(sc.divergence_verdict(div)[0], sc.base.XFAIL) + + def test_module_graph_difference_fails(self): + a = _payload(HASHES, module_graph='{"root":"x"}') + b = _payload(HASHES, module_graph='{"root":"y"}') + div = sc.compare_payloads({"i0": a, "i1": b}) + self.assertTrue(div.module_graph_differs) + self.assertEqual(sc.divergence_verdict(div)[0], sc.base.FAIL) + + def test_module_graph_key_reordering_is_not_a_difference(self): + a = _payload(HASHES, module_graph='{"a":1,"b":2}') + b = _payload(HASHES, module_graph='{"b":2,"a":1}') + div = sc.compare_payloads({"i0": a, "i1": b}) + self.assertFalse(div.module_graph_differs) + self.assertEqual(sc.divergence_verdict(div)[0], sc.base.PASS) + + def test_dep_edge_difference_fails_but_ordering_does_not(self): + a = _payload(HASHES, dep_edges={"//:mid": ["//:core", "//:other"]}) + reordered = _payload(HASHES, dep_edges={"//:mid": ["//:other", "//:core"]}) + different = _payload(HASHES, dep_edges={"//:mid": ["//:core"]}) + self.assertFalse(sc.compare_payloads({"i0": a, "i1": reordered}).dep_edges_differ) + div = sc.compare_payloads({"i0": a, "i1": different}) + self.assertTrue(div.dep_edges_differ) + self.assertEqual(sc.divergence_verdict(div)[0], sc.base.FAIL) + + def test_unparseable_payload_fails_loudly(self): + div = sc.compare_payloads({"i0": _payload(HASHES), "i1": b"not json"}) + self.assertIn("i1", div.parse_errors) + self.assertEqual(sc.divergence_verdict(div)[0], sc.base.FAIL) + + def test_three_instances_report_every_value(self): + a = _payload(HASHES) + b = _payload(dict(HASHES, **{"//:core": "Rule#bbb~bbb"})) + c = _payload(dict(HASHES, **{"//:core": "Rule#ccc~bbb"})) + div = sc.compare_payloads({"i0": a, "i1": b, "i2": c}) + self.assertEqual(len(div.hash_mismatches[0][1]), 3) + + def test_max_labels_caps_the_report_but_not_the_count(self): + many = {f"//:t{i}": f"Rule#{i}~x" for i in range(50)} + other = {f"//:t{i}": f"Rule#{i}~y" for i in range(50)} + div = sc.compare_payloads({"i0": _payload(many), "i1": _payload(other)}, max_labels=5) + self.assertEqual(len(div.hash_mismatches), 5) + self.assertEqual(div.total_hash_mismatches, 50) + + def test_single_body_is_not_a_comparison(self): + div = sc.compare_payloads({"i0": _payload(HASHES)}) + self.assertEqual(div.hash_mismatches, []) + self.assertTrue(div.byte_identical) + + def test_as_json_is_serializable(self): + div = sc.compare_payloads( + {"i0": _payload(HASHES), "i1": _payload(dict(HASHES, **{"//:core": "Rule#z~z"}))} + ) + div.logical_key = "sha.fp" + json.dumps(div.as_json()) # must not raise + + +class FormatTest(unittest.TestCase): + def _diverged(self): + other = dict(HASHES, **{"//:core": "Rule#zzz~bbb", "//:mid": "Rule#yyy~ddd"}) + return sc.compare_payloads({"i0": _payload(HASHES), "i1": _payload(other)}) + + def test_summary_stays_on_one_line(self): + summary = sc.summarize_divergence(self._diverged()) + self.assertNotIn("\n", summary) + self.assertIn("2 label(s) hash differently", summary) + + def test_full_report_lists_each_label(self): + report = sc.format_divergence(self._diverged()) + self.assertIn("//:core: i0=Rule#aaa~bbb i1=Rule#zzz~bbb", report) + self.assertIn("//:mid:", report) + + def test_full_report_notes_what_it_truncated(self): + many = {f"//:t{i}": f"Rule#{i}~x" for i in range(30)} + other = {f"//:t{i}": f"Rule#{i}~y" for i in range(30)} + div = sc.compare_payloads({"i0": _payload(many), "i1": _payload(other)}) + self.assertIn("... 22 more", sc.format_divergence(div, limit=8)) + + def test_identical_payloads_summarize_as_no_differences(self): + div = sc.compare_payloads({"i0": _payload(HASHES), "i1": _payload(HASHES)}) + self.assertEqual(sc.summarize_divergence(div), "no differences") + + +class SkewTest(unittest.TestCase): + def test_clone_names_have_distinct_lengths(self): + names = sc.skewed_clone_names(3) + self.assertEqual(len({len(n) for n in names}), 3) + self.assertEqual(len(set(names)), 3) + + def test_skew_can_be_disabled(self): + names = sc.skewed_clone_names(3, skew=False) + self.assertEqual(len({len(n) for n in names}), 1) + + def test_output_base_is_md5_of_the_absolute_path(self): + import hashlib + from pathlib import Path + + p = Path("/tmp/some/workspace") + self.assertEqual( + sc.output_base_of(p), hashlib.md5(str(p).encode("utf-8")).hexdigest() + ) + + def test_different_paths_give_different_output_bases(self): + from pathlib import Path + + a = sc.output_base_of(Path("/tmp/ws-i0-ppp")) + b = sc.output_base_of(Path("/tmp/ws-i1-pppppppppppppp")) + self.assertNotEqual(a, b) + + +class EnvTest(unittest.TestCase): + def test_aws_env_pins_credentials_and_region(self): + env = sc.aws_env() + # Without credentials the SDK throws SdkClientException, which S3HashCacheStorage swallows + # into a silent miss; without a region createS3Storage() throws and the process dies. + self.assertTrue(env["AWS_ACCESS_KEY_ID"]) + self.assertTrue(env["AWS_SECRET_ACCESS_KEY"]) + self.assertEqual(env["AWS_REGION"], sc.S3_REGION) + + def test_aws_env_neutralizes_ambient_config(self): + env = sc.aws_env() + self.assertIsNone(env["AWS_PROFILE"]) + self.assertNotIn("aws", env["AWS_CONFIG_FILE"].lower()) + self.assertEqual(env["AWS_MAX_ATTEMPTS"], "1") + + def test_instance_flags(self): + inst = sc.Instance(idx=1, iid="i1", clone=None, cache=None) + flags = inst.extra_args("http://127.0.0.1:1234") + self.assertIn("--s3ForcePathStyle", flags) + self.assertEqual(flags[flags.index("--s3Prefix") + 1], "i1/") + self.assertEqual(flags[flags.index("--s3Region") + 1], sc.S3_REGION) + self.assertEqual(flags[flags.index("--s3Endpoint") + 1], "http://127.0.0.1:1234") + + def test_serve_args_are_appended_uniformly(self): + inst = sc.Instance(idx=0, iid="i0", clone=None, cache=None) + flags = inst.extra_args("http://127.0.0.1:1", ["--fineGrainedHashExternalRepos=@x"]) + self.assertEqual(flags[-1], "--fineGrainedHashExternalRepos=@x") + # Still uniform across instances -- several passthrough flags feed the config fingerprint, + # so a per-instance difference would split the fleet across cache keys. + other = sc.Instance(idx=1, iid="i1", clone=None, cache=None) + self.assertEqual( + flags[len(inst.s3_flags("http://127.0.0.1:1")):], + other.extra_args("http://127.0.0.1:1", ["--fineGrainedHashExternalRepos=@x"])[ + len(other.s3_flags("http://127.0.0.1:1")): + ], + ) + + def test_env_skew_is_opt_in(self): + from pathlib import Path + + inst = sc.Instance( + idx=0, iid="i0", clone=None, cache=None, home=Path("/h"), tmpdir=Path("/t") + ) + self.assertNotIn("HOME", inst.env(env_skew=False)) + skewed = inst.env(env_skew=True) + self.assertEqual(skewed["HOME"], "/h") + self.assertEqual(skewed["USER"], "bd-consistency-i0") + + +class ArgParsingTest(unittest.TestCase): + def test_defaults(self): + args = sc.parse_args([]) + self.assertEqual(args.instances, 3) + self.assertFalse(args.env_skew) + self.assertFalse(args.no_path_skew) + self.assertEqual(args.request_timeout, 300.0) + + def test_request_timeout_is_raisable_for_real_repos(self): + self.assertEqual(sc.parse_args(["--request-timeout", "1800"]).request_timeout, 1800.0) + + def test_serve_arg_is_repeatable_and_accepts_flag_shaped_values(self): + args = sc.parse_args( + ["--serve-arg", "--fineGrainedHashExternalRepos=@rules_kotlin", "--serve-arg", "-k"] + ) + self.assertEqual( + args.serve_arg, ["--fineGrainedHashExternalRepos=@rules_kotlin", "-k"] + ) + + def test_flag_shaped_option_value_is_folded(self): + # argparse rejects `--real-clone-args --depth=50` unless it is folded to the `=` spelling; + # the same omission broke the serve-stress cron twice. + args = sc.parse_args(["--real-clone-args", "--depth=50"]) + self.assertEqual(args.real_clone_args, "--depth=50") + + def test_fold_leaves_normal_values_alone(self): + self.assertEqual( + sc.fold_flag_values(["--real-clone-args", "--depth=1", "--instances", "4"]), + ["--real-clone-args=--depth=1", "--instances", "4"], + ) + + def test_real_mode_arguments(self): + args = sc.parse_args( + ["--real-repo-path", "/repo", "--real-from", "abc", "--real-to", "def"] + ) + self.assertEqual(args.real_repo_path, "/repo") + self.assertEqual(args.real_from, "abc") + + def test_every_case_name_is_selectable(self): + for name, _fn in sc.CASES: + selected = [n for n, _ in sc.CASES if n in name] + self.assertIn(name, selected) + + +class SummaryTest(unittest.TestCase): + def _data(self, failures=0, divergences=None): + return { + "meta": { + "instances": 3, + "wall_seconds": 12.3, + "path_skew": True, + "env_skew": False, + }, + "checks": [{"case": "baseline", "name": "ok", "status": "PASS", "detail": "fine"}], + "divergences": divergences or {}, + "s3": { + "run_total": { + "puts": 6, + "gets": 6, + "heads": 0, + "hits": 0, + "objects": 6, + "bad_requests": 0, + } + }, + "output_bases": {"baseline/i0": "abc123"}, + "failures": failures, + } + + def test_clean_run_summary(self): + md = sc.build_summary_md(self._data()) + self.assertIn("hashes are consistent", md) + self.assertIn("| baseline | ok | PASS | fine |", md) + + def test_divergence_is_rendered(self): + divs = { + "baseline": [ + { + "logical_key": "sha.fp", + "total_hash_mismatches": 1, + "label_only_in": {}, + "module_graph_differs": False, + "dep_edges_differ": False, + "hash_mismatches": [ + {"label": "//:core", "hashes": {"i0": "Rule#a~b", "i1": "Rule#z~b"}} + ], + } + ] + } + md = sc.build_summary_md(self._data(failures=1, divergences=divs)) + self.assertIn("Diverging targets", md) + self.assertIn("//:core: i0=Rule#a~b i1=Rule#z~b", md) + + def test_pipes_in_details_do_not_break_the_table(self): + data = self._data() + data["checks"][0]["detail"] = "a | b" + self.assertIn("a \\| b", sc.build_summary_md(data)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/serve_harness.py b/tools/serve_harness.py index a6e0587c..cdffc721 100755 --- a/tools/serve_harness.py +++ b/tools/serve_harness.py @@ -35,6 +35,7 @@ import argparse import contextlib import json +import os import shutil import socket import subprocess @@ -127,6 +128,27 @@ def free_port() -> int: return s.getsockname()[1] +def reserve_ports(n: int) -> list: + """Returns [n] distinct free ports, holding every socket open until all are chosen. + + `free_port()` releases its socket before returning, so calling it n times in a loop can hand + back the same port twice -- harmless for a single server, a real race for a fleet plus a mock + S3 plus a git daemon. Holding the sockets simultaneously makes the numbers distinct; there is + still a window between close and bind, but it is no worse than the single-port case. + """ + socks = [] + try: + for _ in range(n): + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.bind(("127.0.0.1", 0)) + socks.append(s) + return [s.getsockname()[1] for s in socks] + finally: + for s in socks: + with contextlib.suppress(Exception): + s.close() + + def wait_port(port: int, timeout: float) -> bool: deadline = time.time() + timeout while time.time() < deadline: @@ -365,6 +387,10 @@ def serve( track_deps: bool, ready_timeout: float, verbose_server: bool = False, + extra_args=(), + env=None, + port: int | None = None, + bazel: str | None = None, ): """Launches serve, waits for a health verdict, yields (Serve, health) then tears down. @@ -373,8 +399,18 @@ def serve( [verbose_server] launches the subprocess with `-v` so its StderrLogger emits the `[Info]` / `[Warning]` git lines (checkouts, the "cleared stale git index.lock" self-heal) to the captured stderr; without it only `[Error]` lines appear. The stress harness asserts on those lines. + + The remaining options exist for the multi-instance consistency harness, which needs to vary + things this launcher used to hardcode; all four default to today's behaviour exactly. + [extra_args] are appended after the built-in flags (e.g. the `--s3*` remote-cache flags). + [env] is merged over `os.environ` for the child (a skewed HOME/TMPDIR/USER, AWS credentials). + [port] takes a caller-allocated port: `free_port()` closes its socket before returning, so a + fleet allocating N ports in a loop can collide -- the caller can instead hold all N sockets + open at once and pass the numbers here. [bazel] overrides the global for a single instance, + which is how the Bazel-version-skew case runs two versions side by side. """ - port = free_port() + if port is None: + port = free_port() stderr_path = workspace.parent / f"serve.{workspace.name}.stderr.log" args = [ str(LAUNCHER), @@ -385,7 +421,7 @@ def serve( "-w", str(workspace), "-b", - BAZEL, + bazel or BAZEL, "--cacheDir", str(cache), "--port", @@ -395,12 +431,21 @@ def serve( args.append("--no-initial-fetch") if track_deps: args.append("--trackDeps") + args += [str(a) for a in extra_args] + + child_env = None + if env is not None: + child_env = {**os.environ, **{k: str(v) for k, v in env.items() if v is not None}} + for k, v in env.items(): + if v is None: + child_env.pop(k, None) vlog(f"launch serve on :{port} ws={workspace.name} initial_fetch={initial_fetch}") with open(stderr_path, "w") as errf: proc = subprocess.Popen( args, cwd=str(workspace), + env=child_env, stdout=(None if VERBOSE else subprocess.DEVNULL), stderr=(errf if not VERBOSE else None), )