From 806b4405a3df4c2263a5f5c0d18473aac705fa0b Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Wed, 12 Aug 2026 19:02:12 +0000 Subject: [PATCH 1/3] fix(bigtable): resolve bundled accelerator daemon as accelerator.exe on Windows Change-Id: I9f5639b508a7f9e3b0f0599c9692c5ee8bb1db7e --- .../cloud/bigtable/data/_accelerator/_daemon.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py index ead855e21a30..bedd5b112197 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py @@ -41,7 +41,9 @@ _BIN_ENV_VAR = "BIGTABLE_ACCELERATOR_BIN" # Wheels ship the binary at this path relative to the `_accelerator/` package. +# Windows wheels bundle it with a `.exe` suffix (see `_default_binary_path`). _DEFAULT_BIN_RELATIVE_PATH = "bin/accelerator" +_WINDOWS_BIN_SUFFIX = ".exe" # The daemon writes the principal it resolved to this file in its tempdir # (alongside the socket) before binding, so the client can verify it matches @@ -58,8 +60,14 @@ def _default_binary_path() -> str | None: - bundled = os.path.join(os.path.dirname(__file__), _DEFAULT_BIN_RELATIVE_PATH) - return bundled if os.path.isfile(bundled) else None + base = os.path.join(os.path.dirname(__file__), _DEFAULT_BIN_RELATIVE_PATH) + # Windows wheels bundle the daemon as `accelerator.exe`; every other + # platform ships it without a suffix. + candidates = (base + _WINDOWS_BIN_SUFFIX, base) if os.name == "nt" else (base,) + for path in candidates: + if os.path.isfile(path): + return path + return None def _resolve_binary_path() -> str: From 86d5d41272ea256460696a2a36a36501b0e0c2cc Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Tue, 11 Aug 2026 02:27:01 +0000 Subject: [PATCH 2/3] test(bigtable): add accelerator system-test harness and fuzz correctness suite Change-Id: Ie4bb5883f991ba4a7e8b8ce47c12aa77d6d327e6 --- packages/google-cloud-bigtable/noxfile.py | 3 + .../tests/system/data/accelerator/__init__.py | 28 + .../system/data/accelerator/_base_async.py | 142 +++++ .../system/data/accelerator/_base_autogen.py | 131 ++++ .../tests/system/data/accelerator/_harness.py | 602 ++++++++++++++++++ .../tests/system/data/accelerator/conftest.py | 46 ++ .../test_fuzz_correctness_async.py | 128 ++++ .../test_fuzz_correctness_autogen.py | 108 ++++ 8 files changed, 1188 insertions(+) create mode 100644 packages/google-cloud-bigtable/tests/system/data/accelerator/__init__.py create mode 100644 packages/google-cloud-bigtable/tests/system/data/accelerator/_base_async.py create mode 100644 packages/google-cloud-bigtable/tests/system/data/accelerator/_base_autogen.py create mode 100644 packages/google-cloud-bigtable/tests/system/data/accelerator/_harness.py create mode 100644 packages/google-cloud-bigtable/tests/system/data/accelerator/conftest.py create mode 100644 packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_async.py create mode 100644 packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_autogen.py diff --git a/packages/google-cloud-bigtable/noxfile.py b/packages/google-cloud-bigtable/noxfile.py index f8d639a266ed..a909fe54ce2b 100644 --- a/packages/google-cloud-bigtable/noxfile.py +++ b/packages/google-cloud-bigtable/noxfile.py @@ -60,6 +60,9 @@ "pytest-asyncio==0.21.2", RUFF_VERSION, "pyyaml==6.0.2", + # Used by the accelerator pre-release suite for subprocess/FD/tempdir leak + # detection (tests/system/data/accelerator/_harness.py). + "psutil", ] SYSTEM_TEST_LOCAL_DEPENDENCIES: List[str] = [] SYSTEM_TEST_DEPENDENCIES: List[str] = [] diff --git a/packages/google-cloud-bigtable/tests/system/data/accelerator/__init__.py b/packages/google-cloud-bigtable/tests/system/data/accelerator/__init__.py new file mode 100644 index 000000000000..7b02859c0e9d --- /dev/null +++ b/packages/google-cloud-bigtable/tests/system/data/accelerator/__init__.py @@ -0,0 +1,28 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Pre-release integration tests for the Bigtable accelerator. + +Every test in this package exercises the *real* shipped path — the real +``BigtableDataClient`` with ``use_accelerator`` set, the real +``AcceleratorDaemon`` spawning the real bundled Go binary, the real UDS +``_AcceleratorClient``, and (for correctness/stress/backend-error tests) a real +Bigtable instance. Error conditions are induced through real inputs (a bogus +daemon binary, killing the real daemon process, hitting real backend error +conditions), never by substituting a production component with a fake. + +The one place a non-production binary appears is the controlled-binary factory in +``_harness`` (bad / slow daemons). Those are fed as *inputs* to the real +``AcceleratorDaemon`` to deterministically drive its start-failure and +startup-race code paths, which a healthy binary cannot exercise. +""" diff --git a/packages/google-cloud-bigtable/tests/system/data/accelerator/_base_async.py b/packages/google-cloud-bigtable/tests/system/data/accelerator/_base_async.py new file mode 100644 index 000000000000..59508de47431 --- /dev/null +++ b/packages/google-cloud-bigtable/tests/system/data/accelerator/_base_async.py @@ -0,0 +1,142 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Shared base class + fixtures for the accelerator pre-release suite. + +This module is written async-first and converted to a sync twin +(``_base_autogen``) by CrossSync, so async test files and their generated sync +twins can share the exact same fixtures. Everything here drives the *real* +shipped path: ``CrossSync.DataClient(..., use_accelerator=...)`` producing a real +target that spawns the real ``AcceleratorDaemon`` over the real bundled binary. + +The class reuses ``SystemTestRunner`` (temporary instance/table/family creation, +stale-instance cleanup) and adds accelerator-aware client/table fixtures plus a +``janitor`` for per-test row cleanup. +""" + +import os + +import pytest + +from google.cloud.bigtable.data._cross_sync import CrossSync +from google.cloud.bigtable.data.mutations import DeleteAllFromRow, RowMutationEntry + +from . import _harness +from .. import SystemTestRunner + +__CROSS_SYNC_OUTPUT__ = "tests.system.data.accelerator._base_autogen" + + +@CrossSync.convert_class(sync_name="AcceleratorTestBase") +class AcceleratorTestBaseAsync(SystemTestRunner): + """Base for accelerator system tests. + + Subclasses inherit the accelerator-aware fixtures below. Every table fixture + asserts that the accelerator ended up in the expected state (active for + ``accel_table``, native for ``native_table``) so a silent fallback surfaces + as a test failure rather than passing on the wrong code path. + """ + + @pytest.fixture(scope="session", autouse=True) + def _require_accel_env(self): + """Cleanly skip the whole suite when the environment can't support it. + + Runs before the (expensive) instance/table fixtures because it is an + autouse session fixture, so a missing binary or emulator-only setup + skips instead of erroring out mid-provisioning. + """ + _harness.require_real_bigtable_or_skip() + _harness.require_binary_or_skip() + + def _make_client(self, use_accelerator=None): + """Build a real data client with the given accelerator setting.""" + project = os.getenv("GOOGLE_CLOUD_PROJECT") or None + return CrossSync.DataClient(project=project, use_accelerator=use_accelerator) + + def assert_accelerator_active(self, table): + assert table._accelerator_client is not None, ( + "expected the accelerator to be active (use_accelerator=True) but the " + "target fell back to native. Check ADC principal / identity " + "verification and that the bundled daemon binary can start." + ) + + def assert_native(self, table): + assert table._accelerator_client is None, ( + "expected a native target (use_accelerator=False) but an accelerator " + "client was attached." + ) + + @CrossSync.convert + @CrossSync.pytest_fixture(scope="session") + async def client(self): + """Default-on client (``use_accelerator=None``). + + Also backs the ``project_id`` fixture from ``SystemTestRunner``. + """ + async with self._make_client() as client: + yield client + + @CrossSync.convert + @CrossSync.pytest_fixture(scope="session") + async def accel_client(self): + async with self._make_client(use_accelerator=True) as client: + yield client + + @CrossSync.convert + @CrossSync.pytest_fixture(scope="session") + async def native_client(self): + async with self._make_client(use_accelerator=False) as client: + yield client + + @CrossSync.convert + @CrossSync.pytest_fixture(scope="session") + async def accel_table(self, accel_client, instance_id, table_id): + async with accel_client.get_table(instance_id, table_id) as table: + self.assert_accelerator_active(table) + yield table + + @CrossSync.convert + @CrossSync.pytest_fixture(scope="session") + async def native_table(self, native_client, instance_id, table_id): + async with native_client.get_table(instance_id, table_id) as table: + self.assert_native(table) + yield table + + @CrossSync.convert + @CrossSync.pytest_fixture(scope="session") + async def default_table(self, client, instance_id, table_id): + async with client.get_table(instance_id, table_id) as table: + yield table + + @CrossSync.convert + @CrossSync.pytest_fixture(scope="function") + async def janitor(self, native_table): + """Track written row keys and delete them after each test. + + Deletion goes through the native path so cleanup never depends on the + component under test. Yields an object with ``.track(key)``. + """ + + class _Janitor: + def __init__(self): + self.keys = set() + + def track(self, key): + self.keys.add(key) + return key + + j = _Janitor() + yield j + if j.keys: + entries = [RowMutationEntry(key, [DeleteAllFromRow()]) for key in j.keys] + await native_table.bulk_mutate_rows(entries) diff --git a/packages/google-cloud-bigtable/tests/system/data/accelerator/_base_autogen.py b/packages/google-cloud-bigtable/tests/system/data/accelerator/_base_autogen.py new file mode 100644 index 000000000000..a6bd9748ae71 --- /dev/null +++ b/packages/google-cloud-bigtable/tests/system/data/accelerator/_base_autogen.py @@ -0,0 +1,131 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# This file is automatically generated by CrossSync. Do not edit manually. + +"""Shared base class + fixtures for the accelerator pre-release suite. + +This module is written async-first and converted to a sync twin +(``_base_autogen``) by CrossSync, so async test files and their generated sync +twins can share the exact same fixtures. Everything here drives the *real* +shipped path: ``CrossSync.DataClient(..., use_accelerator=...)`` producing a real +target that spawns the real ``AcceleratorDaemon`` over the real bundled binary. + +The class reuses ``SystemTestRunner`` (temporary instance/table/family creation, +stale-instance cleanup) and adds accelerator-aware client/table fixtures plus a +``janitor`` for per-test row cleanup. +""" + +import os + +import pytest + +from google.cloud.bigtable.data._cross_sync import CrossSync +from google.cloud.bigtable.data.mutations import DeleteAllFromRow, RowMutationEntry + +from .. import SystemTestRunner +from . import _harness + + +class AcceleratorTestBase(SystemTestRunner): + """Base for accelerator system tests. + + Subclasses inherit the accelerator-aware fixtures below. Every table fixture + asserts that the accelerator ended up in the expected state (active for + ``accel_table``, native for ``native_table``) so a silent fallback surfaces + as a test failure rather than passing on the wrong code path. + """ + + @pytest.fixture(scope="session", autouse=True) + def _require_accel_env(self): + """Cleanly skip the whole suite when the environment can't support it. + + Runs before the (expensive) instance/table fixtures because it is an + autouse session fixture, so a missing binary or emulator-only setup + skips instead of erroring out mid-provisioning.""" + _harness.require_real_bigtable_or_skip() + _harness.require_binary_or_skip() + + def _make_client(self, use_accelerator=None): + """Build a real data client with the given accelerator setting.""" + project = os.getenv("GOOGLE_CLOUD_PROJECT") or None + return CrossSync._Sync_Impl.DataClient( + project=project, use_accelerator=use_accelerator + ) + + def assert_accelerator_active(self, table): + assert table._accelerator_client is not None, ( + "expected the accelerator to be active (use_accelerator=True) but the target fell back to native. Check ADC principal / identity verification and that the bundled daemon binary can start." + ) + + def assert_native(self, table): + assert table._accelerator_client is None, ( + "expected a native target (use_accelerator=False) but an accelerator client was attached." + ) + + @pytest.fixture(scope="session") + def client(self): + """Default-on client (``use_accelerator=None``). + + Also backs the ``project_id`` fixture from ``SystemTestRunner``.""" + with self._make_client() as client: + yield client + + @pytest.fixture(scope="session") + def accel_client(self): + with self._make_client(use_accelerator=True) as client: + yield client + + @pytest.fixture(scope="session") + def native_client(self): + with self._make_client(use_accelerator=False) as client: + yield client + + @pytest.fixture(scope="session") + def accel_table(self, accel_client, instance_id, table_id): + with accel_client.get_table(instance_id, table_id) as table: + self.assert_accelerator_active(table) + yield table + + @pytest.fixture(scope="session") + def native_table(self, native_client, instance_id, table_id): + with native_client.get_table(instance_id, table_id) as table: + self.assert_native(table) + yield table + + @pytest.fixture(scope="session") + def default_table(self, client, instance_id, table_id): + with client.get_table(instance_id, table_id) as table: + yield table + + @pytest.fixture(scope="function") + def janitor(self, native_table): + """Track written row keys and delete them after each test. + + Deletion goes through the native path so cleanup never depends on the + component under test. Yields an object with ``.track(key)``.""" + + class _Janitor: + def __init__(self): + self.keys = set() + + def track(self, key): + self.keys.add(key) + return key + + j = _Janitor() + yield j + if j.keys: + entries = [RowMutationEntry(key, [DeleteAllFromRow()]) for key in j.keys] + native_table.bulk_mutate_rows(entries) diff --git a/packages/google-cloud-bigtable/tests/system/data/accelerator/_harness.py b/packages/google-cloud-bigtable/tests/system/data/accelerator/_harness.py new file mode 100644 index 000000000000..fc80d1bdec08 --- /dev/null +++ b/packages/google-cloud-bigtable/tests/system/data/accelerator/_harness.py @@ -0,0 +1,602 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Shared, dependency-light helpers for the accelerator pre-release suite. + +This module is intentionally free of async/await and of any Bigtable network IO +so it can be shared verbatim by the async tests, the CrossSync-generated sync +tests, and the standalone stress driver. The pieces are: + +* ``ExpectedState`` — an in-memory model of expected cell state. Every mutation + is applied to the expected state *and* to Bigtable; reads are asserted against + the expected state and cross-checked against the native (non-accelerated) path. +* row/cell normalization + ``assert_rows_equivalent`` for differential checks. +* ``resolve_binary_or_skip`` / ``require_real_bigtable`` gating helpers. +* controlled-binary factories that drive the real daemon's failure/race paths. +* ``ProcessIntrospector`` + ``LeakSnapshot`` (psutil) for subprocess/FD/tempdir + leak detection. +* ``TokenBucket`` and ``LatencyStats`` load-driver primitives (used by the + concurrency tests and the stress driver). +* ``RandomOps`` — a seeded generator of mutations/reads for stress + concurrency. +""" + +from __future__ import annotations + +import glob +import os +import stat +import sys +import time +from dataclasses import dataclass, field +from typing import Any, Callable, Iterable, Sequence + +from google.cloud.bigtable.data import ( + DeleteAllFromFamily, + DeleteAllFromRow, + DeleteRangeFromColumn, + ReadRowsQuery, + SetCell, +) +from google.cloud.bigtable.data.row import Row + +# --------------------------------------------------------------------------- +# Constants shared with the fixtures / tests. +# --------------------------------------------------------------------------- + +# Test tables are created with millisecond timestamp granularity (the Bigtable +# default), so every explicit timestamp must be a whole number of milliseconds. +# The server rejects finer-grained timestamps with InvalidArgument. +MS = 1000 + +# Column families created by the shared system-test conftest. +TEST_FAMILY = "test-family" +TEST_FAMILY_2 = "test-family-2" + +# Env var honored by the real daemon wrapper (google/.../_accelerator/_daemon.py) +# to override the binary location. We use it to point the *real* AcceleratorDaemon +# at a controlled binary for fault injection. +BIN_ENV_VAR = "BIGTABLE_ACCELERATOR_BIN" + + +# --------------------------------------------------------------------------- +# Expected-state model +# --------------------------------------------------------------------------- + +# A single expected cell, in the shape Bigtable returns them. +Expected = tuple # (family: str, qualifier: bytes, timestamp_micros: int, value: bytes) + + +class ExpectedState: + """A minimal, deterministic model of Bigtable cell state. + + Only the semantics the accelerator routes through (``mutate_row`` + + ``read_row``) are modelled: set-cell and the three delete flavors, with + explicit millisecond-granular timestamps so the model is exact. Server-side + timestamps are intentionally out of scope here (they are non-deterministic; + tests that use them assert weaker invariants directly). + + Storage: ``{row_key: {(family, qualifier): {timestamp_micros: value}}}``. + """ + + def __init__(self) -> None: + self._rows: dict[bytes, dict[tuple[str, bytes], dict[int, bytes]]] = {} + + # -- mutation builders: update the model and return the real Mutation ---- + + def set_cell( + self, row_key: bytes, family: str, qualifier: bytes, value: bytes, ts: int + ) -> SetCell: + if ts % MS != 0: + raise ValueError( + "timestamps must be millisecond-granular (multiple of 1000)" + ) + col = self._rows.setdefault(row_key, {}).setdefault((family, qualifier), {}) + col[ts] = value + return SetCell(family, qualifier, value, timestamp_micros=ts) + + def delete_range_from_column( + self, + row_key: bytes, + family: str, + qualifier: bytes, + start: int | None = None, + end: int | None = None, + ) -> DeleteRangeFromColumn: + """Delete cells with start <= ts < end (start=None→0, end=None→inf).""" + col = self._rows.get(row_key, {}).get((family, qualifier)) + if col is not None: + lo = 0 if start is None else start + for ts in list(col): + if ts >= lo and (end is None or ts < end): + del col[ts] + if not col: + self._rows[row_key].pop((family, qualifier), None) + return DeleteRangeFromColumn(family, qualifier, start, end) + + def delete_from_family(self, row_key: bytes, family: str) -> DeleteAllFromFamily: + row = self._rows.get(row_key) + if row is not None: + for key in [k for k in row if k[0] == family]: + del row[key] + return DeleteAllFromFamily(family) + + def delete_from_row(self, row_key: bytes) -> DeleteAllFromRow: + self._rows.pop(row_key, None) + return DeleteAllFromRow() + + # -- expectation ------------------------------------------------------- + + def expected_cells(self, row_key: bytes) -> list[Expected]: + """Return expected cells for a row in Bigtable read order. + + Order: family asc, qualifier asc, timestamp desc — matching the order + the client yields cells (see ``Cell.__lt__``). + """ + row = self._rows.get(row_key, {}) + out: list[Expected] = [] + for family, qualifier in sorted(row, key=lambda k: (k[0], k[1])): + for ts in sorted(row[(family, qualifier)], reverse=True): + out.append((family, qualifier, ts, row[(family, qualifier)][ts])) + return out + + def row_is_empty(self, row_key: bytes) -> bool: + return not self._rows.get(row_key) + + def keys(self) -> Iterable[bytes]: + return list(self._rows) + + +# --------------------------------------------------------------------------- +# Row / cell normalization for differential comparison +# --------------------------------------------------------------------------- + + +def normalize_row(row: Row | None) -> list[Expected]: + """Flatten a ``Row`` into comparable (family, qualifier, ts, value) tuples.""" + if row is None: + return [] + return [ + (cell.family, cell.qualifier, cell.timestamp_micros, cell.value) + for cell in row.cells + ] + + +def _fmt(cells: Sequence[Expected]) -> str: + return ( + "\n".join(f" {fam}:{qual!r}@{ts} = {val!r}" for (fam, qual, ts, val) in cells) + or " " + ) + + +def assert_rows_equivalent( + label_a: str, cells_a: Sequence[Expected], label_b: str, cells_b: Sequence[Expected] +) -> None: + """Assert two normalized rows match, with a readable diff on failure.""" + if list(cells_a) != list(cells_b): + raise AssertionError( + f"row mismatch between {label_a} and {label_b}:\n" + f"{label_a}:\n{_fmt(cells_a)}\n{label_b}:\n{_fmt(cells_b)}" + ) + + +# --------------------------------------------------------------------------- +# Gating helpers +# --------------------------------------------------------------------------- + + +def resolve_binary() -> str | None: + """Return the accelerator binary path if one is available, else None. + + Honors ``BIGTABLE_ACCELERATOR_BIN`` first, then the bundled binary shipped in + the package. Mirrors the daemon wrapper's own resolution so tests skip (not + fail) on platforms where no binary is bundled. + """ + override = os.environ.get(BIN_ENV_VAR) + if override: + return override if os.path.isfile(override) else None + from google.cloud.bigtable.data._accelerator import _daemon + + return _daemon._default_binary_path() + + +def require_binary_or_skip() -> str: + import pytest + + path = resolve_binary() + if path is None: + pytest.skip( + "No accelerator daemon binary available " + f"(set {BIN_ENV_VAR} or install a wheel that bundles it)." + ) + return path + + +def require_real_bigtable_or_skip() -> None: + """Skip when running without real Bigtable credentials/target. + + The accelerator refuses to run against the emulator, so these tests need a + real instance. We treat the presence of the emulator env var as an explicit + "no real backend" signal. + """ + import pytest + + from google.cloud.environment_vars import BIGTABLE_EMULATOR + + if os.environ.get(BIGTABLE_EMULATOR): + pytest.skip("accelerator is not supported against the emulator") + if not (os.environ.get("GOOGLE_CLOUD_PROJECT") or os.environ.get("PROJECT_ID")): + pytest.skip("no GOOGLE_CLOUD_PROJECT set for live accelerator tests") + + +# --------------------------------------------------------------------------- +# Controlled-binary factories (fault + race injection for the real daemon) +# --------------------------------------------------------------------------- + + +def _write_script( + directory: str, name: str, body: str, *, executable: bool = True +) -> str: + path = os.path.join(directory, name) + with open(path, "w") as f: + f.write(body) + if executable: + mode = os.stat(path).st_mode + os.chmod(path, mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) + return path + + +def missing_binary_path(directory: str) -> str: + """A path that does not exist (drives FileNotFoundError resolution).""" + return os.path.join(directory, "does-not-exist-binary") + + +def nonexecutable_binary(directory: str) -> str: + """A real file without the executable bit (drives an OSError on spawn).""" + return _write_script(directory, "not-exec", "#!/bin/sh\nexit 0\n", executable=False) + + +def immediately_exiting_binary(directory: str, exit_code: int = 1) -> str: + """A binary that exits during startup (drives the exit-during-startup path).""" + return _write_script( + directory, + "exits-now", + f"#!/bin/sh\nprintf 'boom\\n' 1>&2\nexit {exit_code}\n", + ) + + +def never_binds_binary(directory: str) -> str: + """A binary that runs, stays alive, but never binds the UDS. + + Drives ``AcceleratorDaemon.start()`` down its startup-timeout path with the + process still alive. Still drive it through ``call_with_timeout`` + + ``force_kill_daemon`` so that a regression of the old ``_read_stderr_tail`` + hang surfaces as a bounded ``TimeoutError`` rather than wedging the suite. + """ + return _write_script(directory, "never-binds", "#!/bin/sh\nexec sleep 3600\n") + + +def slow_bind_binary(directory: str, delay_seconds: float) -> str: + """A binary that binds its UDS after ``delay_seconds``, then serves nothing. + + Used by the startup-race tests to drive ``AcceleratorDaemon._wait_until_ready`` + right up against its timeout. It parses ``--uds-path``, waits, binds an + AF_UNIX socket, and stays alive until stdin closes. + """ + body = f"""#!{sys.executable} +import os, socket, sys, time, signal + +def uds_path(argv): + for i, a in enumerate(argv): + if a == "--uds-path": + return argv[i + 1] + raise SystemExit("no --uds-path") + +path = uds_path(sys.argv) +# Consume the handshake secret line from stdin so the writer never blocks. +try: + sys.stdin.readline() +except Exception: + pass +time.sleep({delay_seconds!r}) +try: + os.unlink(path) +except OSError: + pass +srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) +srv.bind(path) +srv.listen(8) +# Exit when stdin reaches EOF (parent closed it), mirroring the real daemon. +signal.signal(signal.SIGTERM, lambda *a: os._exit(0)) +try: + while True: + line = sys.stdin.readline() + if line == "": + break +except Exception: + pass +os._exit(0) +""" + return _write_script(directory, "slow-bind", body) + + +# --------------------------------------------------------------------------- +# Process / FD / tempdir introspection (leak detection) +# --------------------------------------------------------------------------- + +# Prefix the daemon wrapper uses for its per-daemon tempdir (holds the UDS). +ACCEL_TEMPDIR_GLOB = os.path.join("/tmp", "bt-accel-*") + + +def accel_tempdirs() -> set[str]: + return set(glob.glob(ACCEL_TEMPDIR_GLOB)) + + +@dataclass +class LeakSnapshot: + child_pids: set[int] + num_fds: int + tempdirs: set[str] + + +class ProcessIntrospector: + """Snapshots the current process's children, FDs, and accel tempdirs. + + Requires ``psutil``. Used to assert that constructing and closing accelerated + tables leaves no orphaned daemon subprocesses, leaked file descriptors, or + stray ``/tmp/bt-accel-*`` directories. + """ + + def __init__(self) -> None: + import psutil + + self._psutil = psutil + self._proc = psutil.Process(os.getpid()) + + def snapshot(self) -> LeakSnapshot: + children = set() + for child in self._proc.children(recursive=True): + try: + children.add(child.pid) + except self._psutil.Error: + pass + try: + num_fds = self._proc.num_fds() + except (self._psutil.Error, AttributeError): + num_fds = -1 + return LeakSnapshot( + child_pids=children, num_fds=num_fds, tempdirs=accel_tempdirs() + ) + + def assert_no_leaks( + self, before: LeakSnapshot, *, fd_slack: int = 8, label: str = "" + ) -> None: + after = self.snapshot() + leaked_children = after.child_pids - before.child_pids + leaked_dirs = after.tempdirs - before.tempdirs + prefix = f"[{label}] " if label else "" + assert not leaked_children, ( + f"{prefix}leaked {len(leaked_children)} daemon subprocess(es): " + f"{sorted(leaked_children)}" + ) + assert not leaked_dirs, f"{prefix}leaked accel tempdirs: {sorted(leaked_dirs)}" + if before.num_fds >= 0 and after.num_fds >= 0: + assert after.num_fds <= before.num_fds + fd_slack, ( + f"{prefix}fd count grew from {before.num_fds} to {after.num_fds} " + f"(slack {fd_slack})" + ) + + +def call_with_timeout(fn: Callable[[], Any], timeout: float) -> Any: + """Run ``fn()`` on a daemon thread, raising ``TimeoutError`` if it hangs. + + Used to bound calls into ``AcceleratorDaemon.start()`` so a hang in the + daemon wrapper (see ``never_binds_binary``) cannot wedge the test suite. The + worker thread is left running (it is blocked in a syscall and cannot be + force-joined); callers that time out should ``force_kill_daemon`` to unblock + and reap it. + """ + import threading + + box: dict[str, Any] = {} + + def _run() -> None: + try: + box["result"] = fn() + except BaseException as exc: # noqa: BLE001 - re-raised on the caller thread + box["error"] = exc + + thread = threading.Thread(target=_run, daemon=True) + thread.start() + thread.join(timeout) + if thread.is_alive(): + raise TimeoutError(f"call did not return within {timeout}s") + if "error" in box: + raise box["error"] + return box.get("result") + + +def force_kill_daemon(daemon: Any) -> None: + """Best-effort SIGKILL of a daemon subprocess, ignoring all errors. + + Reaches into the wrapper's private ``_proc`` because the public ``close()`` + can itself block on a wedged process. Safe to call on a half-started daemon. + """ + proc = getattr(daemon, "_proc", None) + if proc is None: + return + try: + proc.kill() + except Exception: + pass + try: + proc.wait(timeout=2.0) + except Exception: + pass + + +def daemon_pid(table: Any) -> int | None: + """Return the real daemon's pid for an accelerated table, or None.""" + daemon = getattr(table, "_accelerator_daemon", None) + if daemon is None: + return None + try: + return daemon.pid + except RuntimeError: + return None + + +def pid_alive(pid: int) -> bool: + import psutil + + try: + proc = psutil.Process(pid) + return proc.is_running() and proc.status() != psutil.STATUS_ZOMBIE + except psutil.Error: + return False + + +# --------------------------------------------------------------------------- +# Load-driver primitives +# --------------------------------------------------------------------------- + + +class TokenBucket: + """A simple monotonic-clock rate limiter shared by concurrency + stress. + + ``time_until_next()`` returns the seconds a caller should sleep before its + next operation to hold the target rate; ``consume()`` records that an + operation happened. Kept pure (clock injected) so it is deterministically + testable and usable from both async and threaded drivers. + """ + + def __init__( + self, rate_per_sec: float, *, clock: Callable[[], float] = time.monotonic + ): + if rate_per_sec <= 0: + raise ValueError("rate_per_sec must be positive") + self._interval = 1.0 / rate_per_sec + self._clock = clock + self._next_at = clock() + + def time_until_next(self) -> float: + return max(0.0, self._next_at - self._clock()) + + def consume(self) -> None: + now = self._clock() + # Advance the schedule; never let it fall arbitrarily behind real time. + self._next_at = max(self._next_at + self._interval, now) + + +@dataclass +class LatencyStats: + """Accumulates latencies (seconds) and computes percentiles on demand.""" + + samples: list[float] = field(default_factory=list) + + def record(self, seconds: float) -> None: + self.samples.append(seconds) + + def percentile(self, pct: float) -> float: + if not self.samples: + return float("nan") + ordered = sorted(self.samples) + k = max( + 0, min(len(ordered) - 1, int(round((pct / 100.0) * (len(ordered) - 1)))) + ) + return ordered[k] + + def summary_ms(self) -> dict[str, float]: + if not self.samples: + return {"count": 0} + return { + "count": len(self.samples), + "p50_ms": self.percentile(50) * 1000, + "p99_ms": self.percentile(99) * 1000, + "max_ms": max(self.samples) * 1000, + "mean_ms": (sum(self.samples) / len(self.samples)) * 1000, + } + + +# --------------------------------------------------------------------------- +# Seeded random operation generator (stress + concurrency) +# --------------------------------------------------------------------------- + + +class RandomOps: + """Deterministic (seeded) generator of keys, values, mutations, and reads. + + Each worker should use a disjoint ``key_prefix`` so concurrent writers never + contend on the same row, keeping the per-worker expected state exact. + """ + + def __init__(self, seed: int, key_prefix: bytes = b"accel-"): + import random + + self._rand = random.Random(seed) + self._prefix = key_prefix + # A small, reused keyspace so reads hit written rows most of the time. + self._keyspace = [key_prefix + f"{i:08d}".encode() for i in range(256)] + + def key(self) -> bytes: + return self._rand.choice(self._keyspace) + + def value(self, max_len: int = 64) -> bytes: + n = self._rand.randint(0, max_len) + return bytes(self._rand.getrandbits(8) for _ in range(n)) + + def ms_timestamp(self) -> int: + # Recent-ish, millisecond-granular timestamps. + return self._rand.randint(1, 2_000_000) * MS + + def family(self) -> str: + return self._rand.choice([TEST_FAMILY, TEST_FAMILY_2]) + + def qualifier(self) -> bytes: + return self._rand.choice([b"q0", b"q1", b"q2", b"q3"]) + + def build_set(self, expected_state: ExpectedState) -> tuple[bytes, SetCell]: + key = self.key() + mut = expected_state.set_cell( + key, self.family(), self.qualifier(), self.value(), self.ms_timestamp() + ) + return key, mut + + def build_mutation(self, expected_state: ExpectedState) -> tuple[bytes, object]: + """Pick a random mutation (weighted toward writes), apply it to the + expected state, and return ``(row_key, mutation)`` to send to the real + table. + + Covers all four accelerator-routed mutation flavors: set-cell and the + three delete kinds, including bounded and open-ended column ranges. + """ + roll = self._rand.random() + key = self.key() + if roll < 0.70: + return key, expected_state.set_cell( + key, self.family(), self.qualifier(), self.value(), self.ms_timestamp() + ) + if roll < 0.82: + # Bounded or half-open column range; keep start <= end when both set. + a = self._rand.choice([None, self.ms_timestamp()]) + b = self._rand.choice([None, self.ms_timestamp()]) + if a is not None and b is not None and a > b: + a, b = b, a + return key, expected_state.delete_range_from_column( + key, self.family(), self.qualifier(), a, b + ) + if roll < 0.92: + return key, expected_state.delete_from_family(key, self.family()) + return key, expected_state.delete_from_row(key) + + def read_query(self, key: bytes) -> ReadRowsQuery: + return ReadRowsQuery(row_keys=key) diff --git a/packages/google-cloud-bigtable/tests/system/data/accelerator/conftest.py b/packages/google-cloud-bigtable/tests/system/data/accelerator/conftest.py new file mode 100644 index 000000000000..ca75018f7f20 --- /dev/null +++ b/packages/google-cloud-bigtable/tests/system/data/accelerator/conftest.py @@ -0,0 +1,46 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Pytest configuration shared by the accelerator system-test package. + +The ``event_loop`` fixture and the async/sync markers are inherited from +``tests/system/conftest.py``. Here we only register the markers this package +uses and provide a scratch directory for the controlled-binary fault-injection +tests. +""" + +import tempfile + +import pytest + + +def pytest_configure(config): + config.addinivalue_line( + "markers", + "accelerator: pre-release integration tests for the Bigtable accelerator", + ) + config.addinivalue_line( + "markers", + "slow: longer-running accelerator tests (leak loops, startup races)", + ) + + +@pytest.fixture +def tmp_bin_dir(): + """A fresh temp directory for building controlled daemon binaries. + + Used by the error-handling and startup tests to hand deliberately broken or + slow binaries to the *real* ``AcceleratorDaemon`` as inputs. + """ + with tempfile.TemporaryDirectory(prefix="accel-test-bin-") as d: + yield d diff --git a/packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_async.py b/packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_async.py new file mode 100644 index 000000000000..71239027a096 --- /dev/null +++ b/packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_async.py @@ -0,0 +1,128 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Randomized differential correctness tests for the accelerator. + +These generate random mutation/read sequences and assert *triple* agreement on +every touched row: + + accelerator read == expected state == native read + +The accelerator read proves the daemon path writes/reads Bigtable correctly; the +expected state proves it matches the intended semantics; the native read proves +the accelerator and the pure-Python client observe identical backend state. + +We drive this with a seeded generator rather than Hypothesis on purpose: every +example issues live RPCs, so shrinking + Hypothesis's 100-example default would +be prohibitively slow and would trip its slow-test health checks. Seeds are +fixed and printed in assertion output, so any failure is fully reproducible. +""" + +import uuid + +import pytest + +from google.cloud.bigtable.data._cross_sync import CrossSync + +from . import _harness + +if CrossSync.is_async: + from ._base_async import AcceleratorTestBaseAsync as AcceleratorTestBase +else: + from ._base_autogen import AcceleratorTestBase + +__CROSS_SYNC_OUTPUT__ = "tests.system.data.accelerator.test_fuzz_correctness_autogen" + + +@CrossSync.convert_class(sync_name="TestFuzzCorrectness") +class TestFuzzCorrectnessAsync(AcceleratorTestBase): + """Randomized + targeted correctness of the real accelerator write/read path.""" + + NUM_OPS = 40 + + @CrossSync.convert + async def _verify_key(self, accel_table, native_table, expected_state, key): + """Assert accelerator read == expected == native read for one row.""" + accel_row = await accel_table.read_row(key) + native_row = await native_table.read_row(key) + expected = expected_state.expected_cells(key) + accel_cells = _harness.normalize_row(accel_row) + native_cells = _harness.normalize_row(native_row) + _harness.assert_rows_equivalent( + "accelerator", accel_cells, "expected", expected + ) + _harness.assert_rows_equivalent( + "accelerator", accel_cells, "native", native_cells + ) + + @pytest.mark.parametrize("seed", [0, 1, 2]) + @CrossSync.pytest + async def test_random_mutations_match_model_and_native( + self, accel_table, native_table, janitor, seed + ): + """Apply a random mix of set/delete mutations through the accelerator and + continuously reconcile against the expected state and the native client.""" + # A unique key prefix per run keeps each expected state exact even if a + # previous run's cleanup was incomplete (no cross-test row contamination). + prefix = f"fuzz-{seed}-{uuid.uuid4().hex[:8]}-".encode() + ops = _harness.RandomOps(seed, key_prefix=prefix) + expected_state = _harness.ExpectedState() + touched: set[bytes] = set() + + for i in range(self.NUM_OPS): + key, mutation = ops.build_mutation(expected_state) + await accel_table.mutate_row(key, mutation) + janitor.track(key) + touched.add(key) + # Reconcile a random already-touched row part-way through, so ordering + # bugs surface mid-sequence rather than only at the end. + if touched and i % 5 == 4: + vkey = ops._rand.choice(sorted(touched)) + await self._verify_key(accel_table, native_table, expected_state, vkey) + + # Final full reconciliation of every row we touched. + for key in sorted(touched): + await self._verify_key(accel_table, native_table, expected_state, key) + + @CrossSync.pytest + async def test_multiple_versions_and_range_delete( + self, accel_table, native_table, janitor + ): + """Deterministic coverage of multi-version cells + a bounded range delete, + so this behavior is exercised regardless of the random draw.""" + from google.cloud.bigtable.data.mutations import DeleteRangeFromColumn + + key = janitor.track(f"versions-{uuid.uuid4().hex}".encode()) + family = _harness.TEST_FAMILY + qualifier = b"q0" + expected_state = _harness.ExpectedState() + + # Three explicit versions of the same cell. + for ts_ms, value in [(1000, b"v1"), (2000, b"v2"), (3000, b"v3")]: + ts = ts_ms * _harness.MS + await accel_table.mutate_row( + key, expected_state.set_cell(key, family, qualifier, value, ts) + ) + await self._verify_key(accel_table, native_table, expected_state, key) + + # Delete the middle version only: [2_000_000, 3_000_000) removes v2@2000. + expected_state.delete_range_from_column( + key, family, qualifier, 2000 * _harness.MS, 3000 * _harness.MS + ) + await accel_table.mutate_row( + key, + DeleteRangeFromColumn( + family, qualifier, 2000 * _harness.MS, 3000 * _harness.MS + ), + ) + await self._verify_key(accel_table, native_table, expected_state, key) diff --git a/packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_autogen.py b/packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_autogen.py new file mode 100644 index 000000000000..bc232433aa2b --- /dev/null +++ b/packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_autogen.py @@ -0,0 +1,108 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# This file is automatically generated by CrossSync. Do not edit manually. + +"""Randomized differential correctness tests for the accelerator. + +These generate random mutation/read sequences and assert *triple* agreement on +every touched row: + + accelerator read == expected state == native read + +The accelerator read proves the daemon path writes/reads Bigtable correctly; the +expected state proves it matches the intended semantics; the native read proves +the accelerator and the pure-Python client observe identical backend state. + +We drive this with a seeded generator rather than Hypothesis on purpose: every +example issues live RPCs, so shrinking + Hypothesis's 100-example default would +be prohibitively slow and would trip its slow-test health checks. Seeds are +fixed and printed in assertion output, so any failure is fully reproducible. +""" + +import uuid + +import pytest + +from . import _harness +from ._base_autogen import AcceleratorTestBase + + +class TestFuzzCorrectness(AcceleratorTestBase): + """Randomized + targeted correctness of the real accelerator write/read path.""" + + NUM_OPS = 40 + + def _verify_key(self, accel_table, native_table, expected_state, key): + """Assert accelerator read == expected == native read for one row.""" + accel_row = accel_table.read_row(key) + native_row = native_table.read_row(key) + expected = expected_state.expected_cells(key) + accel_cells = _harness.normalize_row(accel_row) + native_cells = _harness.normalize_row(native_row) + _harness.assert_rows_equivalent( + "accelerator", accel_cells, "expected", expected + ) + _harness.assert_rows_equivalent( + "accelerator", accel_cells, "native", native_cells + ) + + @pytest.mark.parametrize("seed", [0, 1, 2]) + def test_random_mutations_match_model_and_native( + self, accel_table, native_table, janitor, seed + ): + """Apply a random mix of set/delete mutations through the accelerator and + continuously reconcile against the expected state and the native client.""" + prefix = f"fuzz-{seed}-{uuid.uuid4().hex[:8]}-".encode() + ops = _harness.RandomOps(seed, key_prefix=prefix) + expected_state = _harness.ExpectedState() + touched: set[bytes] = set() + for i in range(self.NUM_OPS): + key, mutation = ops.build_mutation(expected_state) + accel_table.mutate_row(key, mutation) + janitor.track(key) + touched.add(key) + if touched and i % 5 == 4: + vkey = ops._rand.choice(sorted(touched)) + self._verify_key(accel_table, native_table, expected_state, vkey) + for key in sorted(touched): + self._verify_key(accel_table, native_table, expected_state, key) + + def test_multiple_versions_and_range_delete( + self, accel_table, native_table, janitor + ): + """Deterministic coverage of multi-version cells + a bounded range delete, + so this behavior is exercised regardless of the random draw.""" + from google.cloud.bigtable.data.mutations import DeleteRangeFromColumn + + key = janitor.track(f"versions-{uuid.uuid4().hex}".encode()) + family = _harness.TEST_FAMILY + qualifier = b"q0" + expected_state = _harness.ExpectedState() + for ts_ms, value in [(1000, b"v1"), (2000, b"v2"), (3000, b"v3")]: + ts = ts_ms * _harness.MS + accel_table.mutate_row( + key, expected_state.set_cell(key, family, qualifier, value, ts) + ) + self._verify_key(accel_table, native_table, expected_state, key) + expected_state.delete_range_from_column( + key, family, qualifier, 2000 * _harness.MS, 3000 * _harness.MS + ) + accel_table.mutate_row( + key, + DeleteRangeFromColumn( + family, qualifier, 2000 * _harness.MS, 3000 * _harness.MS + ), + ) + self._verify_key(accel_table, native_table, expected_state, key) From 144863953d8303254912e3bbdb4eab9de1db8603 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Wed, 12 Aug 2026 13:57:39 +0000 Subject: [PATCH 3/3] test(bigtable): fuzz aggregate add-to-cell, wider column spread, 100k ops Change-Id: I20f6a95a1da2f66d6413e8c2f4344b4771ddc220 --- .../tests/system/data/accelerator/_harness.py | 120 +++++++++++++++--- .../test_fuzz_correctness_async.py | 11 +- .../test_fuzz_correctness_autogen.py | 7 +- 3 files changed, 116 insertions(+), 22 deletions(-) diff --git a/packages/google-cloud-bigtable/tests/system/data/accelerator/_harness.py b/packages/google-cloud-bigtable/tests/system/data/accelerator/_harness.py index fc80d1bdec08..0d83b96a40c1 100644 --- a/packages/google-cloud-bigtable/tests/system/data/accelerator/_harness.py +++ b/packages/google-cloud-bigtable/tests/system/data/accelerator/_harness.py @@ -41,6 +41,7 @@ from typing import Any, Callable, Iterable, Sequence from google.cloud.bigtable.data import ( + AddToCell, DeleteAllFromFamily, DeleteAllFromRow, DeleteRangeFromColumn, @@ -61,6 +62,12 @@ # Column families created by the shared system-test conftest. TEST_FAMILY = "test-family" TEST_FAMILY_2 = "test-family-2" +# An int64 "sum" aggregate family (see ``column_family_config`` in the system +# conftest). Only ``AddToCell`` may write to it; ``SetCell`` is server-rejected. +# Cells read back as 8-byte big-endian signed int64 of the accumulated sum. +TEST_AGGREGATE_FAMILY = "test-aggregate-family" +# Width of an aggregate cell's value on read (int64, big-endian, signed). +_AGGREGATE_VALUE_BYTES = 8 # Env var honored by the real daemon wrapper (google/.../_accelerator/_daemon.py) # to override the binary location. We use it to point the *real* AcceleratorDaemon @@ -80,12 +87,16 @@ class ExpectedState: """A minimal, deterministic model of Bigtable cell state. Only the semantics the accelerator routes through (``mutate_row`` + - ``read_row``) are modelled: set-cell and the three delete flavors, with - explicit millisecond-granular timestamps so the model is exact. Server-side - timestamps are intentionally out of scope here (they are non-deterministic; - tests that use them assert weaker invariants directly). - - Storage: ``{row_key: {(family, qualifier): {timestamp_micros: value}}}``. + ``read_row``) are modelled: set-cell, add-to-cell (int64 ``sum`` aggregate), + and the three delete flavors, with explicit millisecond-granular timestamps + so the model is exact. Server-side timestamps are intentionally out of scope + here (they are non-deterministic; tests that use them assert weaker + invariants directly). + + Storage: ``{row_key: {(family, qualifier): {timestamp_micros: value}}}``. For + ordinary families ``value`` is ``bytes``; for ``TEST_AGGREGATE_FAMILY`` it is + the accumulated ``int`` sum, encoded to 8-byte big-endian at read time so it + matches what Bigtable returns for an int64 aggregate cell. """ def __init__(self) -> None: @@ -104,6 +115,29 @@ def set_cell( col[ts] = value return SetCell(family, qualifier, value, timestamp_micros=ts) + def add_to_cell( + self, row_key: bytes, qualifier: bytes, delta: int, ts: int + ) -> AddToCell: + """Accumulate ``delta`` into an int64 ``sum`` aggregate cell. + + Aggregate cells live only in ``TEST_AGGREGATE_FAMILY``. Repeated adds at + the same (qualifier, timestamp) sum server-side — this mutation is *not* + idempotent, which is exactly why the model tracks the running total + rather than the last write. The stored value is an ``int``; it is encoded + to big-endian bytes in ``expected_cells``. + """ + if ts % MS != 0: + raise ValueError( + "timestamps must be millisecond-granular (multiple of 1000)" + ) + col = self._rows.setdefault(row_key, {}).setdefault( + (TEST_AGGREGATE_FAMILY, qualifier), {} + ) + col[ts] = col.get(ts, 0) + delta + return AddToCell( + TEST_AGGREGATE_FAMILY, qualifier, delta, timestamp_micros=ts + ) + def delete_range_from_column( self, row_key: bytes, @@ -146,7 +180,14 @@ def expected_cells(self, row_key: bytes) -> list[Expected]: out: list[Expected] = [] for family, qualifier in sorted(row, key=lambda k: (k[0], k[1])): for ts in sorted(row[(family, qualifier)], reverse=True): - out.append((family, qualifier, ts, row[(family, qualifier)][ts])) + value = row[(family, qualifier)][ts] + if family == TEST_AGGREGATE_FAMILY: + # Bigtable returns an int64 aggregate cell as 8 big-endian, + # signed bytes; mirror that so the differential check matches. + value = int(value).to_bytes( + _AGGREGATE_VALUE_BYTES, "big", signed=True + ) + out.append((family, qualifier, ts, value)) return out def row_is_empty(self, row_key: bytes) -> bool: @@ -539,13 +580,35 @@ class RandomOps: contend on the same row, keeping the per-worker expected state exact. """ - def __init__(self, seed: int, key_prefix: bytes = b"accel-"): + def __init__( + self, + seed: int, + key_prefix: bytes = b"accel-", + *, + include_aggregate: bool = False, + ): import random self._rand = random.Random(seed) self._prefix = key_prefix # A small, reused keyspace so reads hit written rows most of the time. self._keyspace = [key_prefix + f"{i:08d}".encode() for i in range(256)] + # Aggregate ops need ``TEST_AGGREGATE_FAMILY`` on the table; keep them + # opt-in so callers running against tables without that family (e.g. the + # self-managed stress table) are unaffected. + self._include_aggregate = include_aggregate + # Randomize the column cardinality per instance (seeded) so different + # runs spread cells across a variable number of qualifiers rather than a + # fixed handful of columns. The number of *families* is capped by the + # table schema: two ordinary families, plus the aggregate family when + # enabled. + self._families = [TEST_FAMILY, TEST_FAMILY_2] + self._qualifiers = [ + f"q{i}".encode() for i in range(self._rand.randint(2, 16)) + ] + self._agg_qualifiers = [ + f"agg{i}".encode() for i in range(self._rand.randint(1, 8)) + ] def key(self) -> bytes: return self._rand.choice(self._keyspace) @@ -559,10 +622,30 @@ def ms_timestamp(self) -> int: return self._rand.randint(1, 2_000_000) * MS def family(self) -> str: - return self._rand.choice([TEST_FAMILY, TEST_FAMILY_2]) + return self._rand.choice(self._families) def qualifier(self) -> bytes: - return self._rand.choice([b"q0", b"q1", b"q2", b"q3"]) + return self._rand.choice(self._qualifiers) + + def agg_qualifier(self) -> bytes: + return self._rand.choice(self._agg_qualifiers) + + def add_delta(self) -> int: + # Bounded so even a long run of adds to one cell stays well inside int64. + return self._rand.randint(-(2**20), 2**20) + + def _delete_target(self) -> tuple[str, bytes]: + """A coherent (family, qualifier) to delete from — sometimes the + aggregate family so aggregate cells are actually cleared, not just set.""" + if self._include_aggregate and self._rand.random() < 0.30: + return TEST_AGGREGATE_FAMILY, self.agg_qualifier() + return self.family(), self.qualifier() + + def _delete_family(self) -> str: + families = self._families + ( + [TEST_AGGREGATE_FAMILY] if self._include_aggregate else [] + ) + return self._rand.choice(families) def build_set(self, expected_state: ExpectedState) -> tuple[bytes, SetCell]: key = self.key() @@ -576,11 +659,17 @@ def build_mutation(self, expected_state: ExpectedState) -> tuple[bytes, object]: expected state, and return ``(row_key, mutation)`` to send to the real table. - Covers all four accelerator-routed mutation flavors: set-cell and the - three delete kinds, including bounded and open-ended column ranges. + Covers every accelerator-routed mutation flavor: set-cell, the three + delete kinds (bounded and open-ended column ranges included), and — when + ``include_aggregate`` is set — the non-idempotent int64 ``sum`` + add-to-cell aggregate. """ roll = self._rand.random() key = self.key() + if self._include_aggregate and roll < 0.18: + return key, expected_state.add_to_cell( + key, self.agg_qualifier(), self.add_delta(), self.ms_timestamp() + ) if roll < 0.70: return key, expected_state.set_cell( key, self.family(), self.qualifier(), self.value(), self.ms_timestamp() @@ -591,11 +680,10 @@ def build_mutation(self, expected_state: ExpectedState) -> tuple[bytes, object]: b = self._rand.choice([None, self.ms_timestamp()]) if a is not None and b is not None and a > b: a, b = b, a - return key, expected_state.delete_range_from_column( - key, self.family(), self.qualifier(), a, b - ) + fam, qual = self._delete_target() + return key, expected_state.delete_range_from_column(key, fam, qual, a, b) if roll < 0.92: - return key, expected_state.delete_from_family(key, self.family()) + return key, expected_state.delete_from_family(key, self._delete_family()) return key, expected_state.delete_from_row(key) def read_query(self, key: bytes) -> ReadRowsQuery: diff --git a/packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_async.py b/packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_async.py index 71239027a096..36f7f8900bde 100644 --- a/packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_async.py +++ b/packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_async.py @@ -48,7 +48,11 @@ class TestFuzzCorrectnessAsync(AcceleratorTestBase): """Randomized + targeted correctness of the real accelerator write/read path.""" - NUM_OPS = 40 + # High op count so the seeded differential sweep exercises a large, + # realistic mix of mutations against the real backend. This is a live, + # opt-in system test (skipped without real-Bigtable creds), not a CI unit + # test, so the RPC volume is intentional. + NUM_OPS = 100_000 @CrossSync.convert async def _verify_key(self, accel_table, native_table, expected_state, key): @@ -70,12 +74,13 @@ async def _verify_key(self, accel_table, native_table, expected_state, key): async def test_random_mutations_match_model_and_native( self, accel_table, native_table, janitor, seed ): - """Apply a random mix of set/delete mutations through the accelerator and + """Apply a random mix of set/add/delete mutations (including the + non-idempotent int64 aggregate add-to-cell) through the accelerator and continuously reconcile against the expected state and the native client.""" # A unique key prefix per run keeps each expected state exact even if a # previous run's cleanup was incomplete (no cross-test row contamination). prefix = f"fuzz-{seed}-{uuid.uuid4().hex[:8]}-".encode() - ops = _harness.RandomOps(seed, key_prefix=prefix) + ops = _harness.RandomOps(seed, key_prefix=prefix, include_aggregate=True) expected_state = _harness.ExpectedState() touched: set[bytes] = set() diff --git a/packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_autogen.py b/packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_autogen.py index bc232433aa2b..45d172fbe8b2 100644 --- a/packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_autogen.py +++ b/packages/google-cloud-bigtable/tests/system/data/accelerator/test_fuzz_correctness_autogen.py @@ -42,7 +42,7 @@ class TestFuzzCorrectness(AcceleratorTestBase): """Randomized + targeted correctness of the real accelerator write/read path.""" - NUM_OPS = 40 + NUM_OPS = 100000 def _verify_key(self, accel_table, native_table, expected_state, key): """Assert accelerator read == expected == native read for one row.""" @@ -62,10 +62,11 @@ def _verify_key(self, accel_table, native_table, expected_state, key): def test_random_mutations_match_model_and_native( self, accel_table, native_table, janitor, seed ): - """Apply a random mix of set/delete mutations through the accelerator and + """Apply a random mix of set/add/delete mutations (including the + non-idempotent int64 aggregate add-to-cell) through the accelerator and continuously reconcile against the expected state and the native client.""" prefix = f"fuzz-{seed}-{uuid.uuid4().hex[:8]}-".encode() - ops = _harness.RandomOps(seed, key_prefix=prefix) + ops = _harness.RandomOps(seed, key_prefix=prefix, include_aggregate=True) expected_state = _harness.ExpectedState() touched: set[bytes] = set() for i in range(self.NUM_OPS):