diff --git a/.github/workflows/nightly_run.yml b/.github/workflows/nightly_run.yml index 228b0b7de22..43ae2dcf414 100644 --- a/.github/workflows/nightly_run.yml +++ b/.github/workflows/nightly_run.yml @@ -45,6 +45,55 @@ jobs: # to avoid OOM issues (these tests use >5GiB memory each) cargo test --release --package lance-encoding -- --ignored jumbo --test-threads=1 + commit-conformance-fault-sweep: + if: github.repository == 'lance-format/lance' + timeout-minutes: 180 + runs-on: ubuntu-24.04-8x + defaults: + run: + shell: bash + working-directory: python + steps: + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 + with: + fetch-depth: 0 + lfs: true + - name: Set up Python + uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: "3.13" + - uses: actions-rust-lang/setup-rust-toolchain@a0b538fa0b742a6aa35d6e2c169b4bd06d225a98 # v1 + - uses: Swatinem/rust-cache@779680da715d629ac1d338a641029a2f4372abb5 # v2 + with: + workspaces: python + - name: Set up uv + uses: astral-sh/setup-uv@d0cc045d04ccac9d8b7881df0226f9e82c39688e # v6 + - name: Install build dependencies + run: | + sudo apt update + sudo apt install -y protobuf-compiler libssl-dev + - name: Build Lance + run: make install + - name: Start S3-compatible services + working-directory: . + run: docker compose -f docker-compose.yml --profile commit-conformance up -d --wait + - name: Run local commit conformance sweep + env: + LANCE_COMMIT_CONFORMANCE_TRACE_DIR: target/commit-conformance-traces + run: uv run pytest --run-integration -m "not real_s3" -q python/tests/test_commit_conformance.py + - name: Upload commit conformance traces + if: always() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 + with: + name: nightly-commit-conformance-traces + path: python/target/commit-conformance-traces + if-no-files-found: ignore + retention-days: 14 + - name: Stop S3-compatible services + if: always() + working-directory: . + run: docker compose -f docker-compose.yml --profile commit-conformance down -v --remove-orphans --timeout 0 + # Cross-version index maintenance-sequence search (see python/tests/compat/compat_sequence.py). # Ages an index under the latest release of each of the two previous majors and exercises it # under this commit, searching op sequences for panics or correctness divergence. The search is diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml index 98831ac6648..68df268873c 100644 --- a/.github/workflows/python.yml +++ b/.github/workflows/python.yml @@ -18,6 +18,10 @@ on: - .github/workflows/build_linux_wheel/** - .github/workflows/build_mac_wheel/** - .github/workflows/run_tests/** + - .github/workflows/run_integtests/** + - .github/workflows/nightly_run.yml + - ci/run_real_s3_commit_conformance.sh + - docker-compose.yml permissions: contents: read @@ -320,3 +324,11 @@ jobs: pip install ray[data] pip install torch --index-url https://download.pytorch.org/whl/cpu - uses: ./.github/workflows/run_integtests + - name: Upload commit conformance traces + if: always() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 + with: + name: commit-conformance-traces + path: python/target/commit-conformance-traces + if-no-files-found: ignore + retention-days: 7 diff --git a/.github/workflows/run_integtests/action.yml b/.github/workflows/run_integtests/action.yml index 38115e49fea..8c36f0115c0 100644 --- a/.github/workflows/run_integtests/action.yml +++ b/.github/workflows/run_integtests/action.yml @@ -9,10 +9,10 @@ runs: shell: bash run: | pip3 install $(ls target/wheels/pylance-*.whl)[tests,ray] - - name: Start localstack + - name: Start S3-compatible services shell: bash run: | - docker compose -f docker-compose.yml up -d --wait + docker compose -f docker-compose.yml --profile commit-conformance up -d --wait - name: Run python tests shell: bash working-directory: python diff --git a/ci/run_real_s3_commit_conformance.sh b/ci/run_real_s3_commit_conformance.sh new file mode 100755 index 00000000000..51b24fa1adc --- /dev/null +++ b/ci/run_real_s3_commit_conformance.sh @@ -0,0 +1,110 @@ +#!/usr/bin/env bash +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The Lance Authors + +set -euo pipefail + +if [[ -z "${LANCE_CONFORMANCE_REAL_S3_BUCKET:-}" ]]; then + echo "LANCE_CONFORMANCE_REAL_S3_BUCKET must name an existing test bucket" >&2 + exit 2 +fi + +for command in flock getent iptables setsid sudo; do + if ! command -v "${command}" >/dev/null 2>&1; then + echo "${command} is required by the fail-closed real-S3 runner" >&2 + exit 2 + fi +done +sudo -n true + +region="${LANCE_CONFORMANCE_REAL_S3_REGION:-us-east-2}" +proxy_port="${LANCE_CONFORMANCE_PROXY_PORT:-18080}" +s3_host="s3.${region}.amazonaws.com" +test_user_id="$(id -u)" +nat_chain="LANCE_CC_S3" +guard_chain="LANCE_CC_S3_GUARD" +hosts_marker="# lance-commit-conformance" +lock_file="/tmp/lance-commit-conformance.lock" +test_pid="" + +exec 9>"${lock_file}" +if ! flock -n 9; then + echo "another real-S3 commit conformance run holds ${lock_file}" >&2 + exit 2 +fi + +remove_stale_network_state() { + sudo iptables -t nat -D OUTPUT \ + -p tcp -d 127.0.0.1/32 --dport 80 \ + -m owner --uid-owner "${test_user_id}" -j "${nat_chain}" \ + 2>/dev/null || true + sudo iptables -D OUTPUT \ + -p tcp -d 127.0.0.1/32 --dport 80 \ + -m owner --uid-owner "${test_user_id}" -j "${guard_chain}" \ + 2>/dev/null || true + sudo iptables -t nat -F "${nat_chain}" 2>/dev/null || true + sudo iptables -t nat -X "${nat_chain}" 2>/dev/null || true + sudo iptables -F "${guard_chain}" 2>/dev/null || true + sudo iptables -X "${guard_chain}" 2>/dev/null || true + sudo sed -i "\|${hosts_marker}$|d" /etc/hosts +} + +cleanup() { + status=$? + trap - EXIT INT TERM + set +e + if [[ -n "${test_pid}" ]] && kill -0 "${test_pid}" 2>/dev/null; then + kill -TERM -- "-${test_pid}" 2>/dev/null || true + for _ in {1..20}; do + kill -0 "${test_pid}" 2>/dev/null || break + sleep 0.1 + done + kill -KILL -- "-${test_pid}" 2>/dev/null || true + wait "${test_pid}" 2>/dev/null || true + fi + remove_stale_network_state + if grep -Fq "${hosts_marker}" /etc/hosts; then + echo "failed to remove the real-S3 /etc/hosts isolation entry" >&2 + status=1 + fi + exit "${status}" +} +trap cleanup EXIT +trap 'exit 130' INT +trap 'exit 143' TERM + +# Recover exact state left by a previously killed runner before resolving the +# real upstream address. The generated /etc/hosts entry maps only the regional +# S3 hostname and is tagged with hosts_marker. +remove_stale_network_state +upstream_ip="$(getent ahostsv4 "${s3_host}" | awk 'NR == 1 { print $1 }')" +if [[ -z "${upstream_ip}" || "${upstream_ip}" == "127.0.0.1" ]]; then + echo "could not resolve a non-loopback IPv4 address for ${s3_host}" >&2 + exit 2 +fi + +# Resolve the test client's S3 hostname to loopback. Its only port-80 route is +# redirected to the fault proxy. The filter guard rejects the request if the +# NAT redirect disappears, so signed plaintext cannot escape to real S3. +printf '127.0.0.1 %s %s\n' "${s3_host}" "${hosts_marker}" | sudo tee -a /etc/hosts >/dev/null +sudo iptables -t nat -N "${nat_chain}" +sudo iptables -t nat -A "${nat_chain}" \ + -p tcp -j REDIRECT --to-ports "${proxy_port}" +sudo iptables -t nat -I OUTPUT 1 \ + -p tcp -d 127.0.0.1/32 --dport 80 \ + -m owner --uid-owner "${test_user_id}" -j "${nat_chain}" +sudo iptables -N "${guard_chain}" +sudo iptables -A "${guard_chain}" -j REJECT +sudo iptables -I OUTPUT 1 \ + -p tcp -d 127.0.0.1/32 --dport 80 \ + -m owner --uid-owner "${test_user_id}" -j "${guard_chain}" + +cd "$(dirname "$0")/../python" +export LANCE_COMMIT_CONFORMANCE_TRACE_DIR="${LANCE_COMMIT_CONFORMANCE_TRACE_DIR:-target/commit-conformance-traces}" +export LANCE_CONFORMANCE_REAL_S3_ISOLATED=1 +export LANCE_CONFORMANCE_REAL_S3_UPSTREAM_IP="${upstream_ip}" +setsid uv run --frozen pytest --run-integration -m "recurring and real_s3" -q \ + python/tests/test_commit_conformance.py & +test_pid=$! +wait "${test_pid}" +test_pid="" diff --git a/docker-compose.yml b/docker-compose.yml index 6b87efc58dd..fa6deb642c8 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,9 +1,8 @@ -version: "3.9" services: localstack: image: localstack/localstack:4.0 ports: - - 4566:4566 + - 127.0.0.1:4566:4566 environment: - SERVICES=s3,dynamodb,kms - DOCKER_HOST=unix:///var/run/docker.sock @@ -15,3 +14,17 @@ services: interval: 5s retries: 3 start_period: 10s + minio: + image: minio/minio:RELEASE.2025-09-07T16-13-09Z + profiles: ["commit-conformance"] + command: server /data + ports: + - 127.0.0.1:9000:9000 + environment: + - MINIO_ROOT_USER=ACCESS_KEY + - MINIO_ROOT_PASSWORD=SECRET_KEY + healthcheck: + test: [ "CMD", "curl", "-f", "http://localhost:9000/minio/health/live" ] + interval: 5s + retries: 3 + start_period: 10s diff --git a/python/DEVELOPMENT.md b/python/DEVELOPMENT.md index 18e1b9b1eef..a433cefccdf 100644 --- a/python/DEVELOPMENT.md +++ b/python/DEVELOPMENT.md @@ -267,19 +267,45 @@ sophisticated enough to represent asynchronous parallel work. As a result, a single instrumented async method may appear as many different spans in the UI. -## Running S3 Integration tests +## Running S3 integration tests -The integration tests run against local minio and local dynamodb. To start the -services, run +The integration tests run against MinIO and LocalStack DynamoDB. The standard +entry point starts both services, runs the integration tests, and tears the +services down: ```shell -docker compose up +uv run make integtest ``` -Then you can run the tests with +The PR commit-conformance matrix covers append, compaction, add-columns, and +create-index through both conditional object-store and DynamoDB external +metadata commits. Each operation is checked under lost commit responses, lost +verification responses, and a deterministic foreign-writer race. The oracle +checks the complete visible version history, the latest-opened version, unique +transaction identities, and the correspondence between successful writer +results and persisted transactions. Request traces include executable fault +plans and are written to `target/commit-conformance-traces`. + +The complete fault sweep is marked `recurring` and runs nightly. To run it +locally against already-started services: + +```shell +uv run pytest --run-integration -m "recurring and not real_s3" \ + python/tests/test_commit_conformance.py +``` + +Real S3 response-loss validation requires a Linux host with AWS credentials and +an existing test bucket. Always use the runner: the tests refuse to send real-S3 +traffic unless it has installed the fail-closed network guard. The runner maps +only the regional S3 hostname to loopback, redirects that loopback traffic to +the fault proxy, and rejects it if the redirect is missing. The proxy connects +to a pre-resolved S3 address over HTTPS while preserving TLS hostname +verification. A lock and exact cleanup markers make concurrent or stale runs +detectable. ```shell -pytest --run-integration python/tests/test_s3_ddb.py +LANCE_CONFORMANCE_REAL_S3_BUCKET=my-test-bucket \ + ../ci/run_real_s3_commit_conformance.sh ``` ## Building wheels locally diff --git a/python/Makefile b/python/Makefile index d5077019f35..e1ebfd40174 100644 --- a/python/Makefile +++ b/python/Makefile @@ -4,6 +4,8 @@ PYTHON ?= PYTEST_ARGS ?= -vvv -s -m "not recurring" KEEP_COMPOSE ?= 0 COMPOSE_FILE ?= ../docker-compose.yml +COMPOSE_PROFILE ?= commit-conformance +COMMIT_CONFORMANCE_TRACE_DIR ?= target/commit-conformance-traces UV_SYNC = uv sync UV_RUN = uv run --frozen @@ -35,14 +37,14 @@ build: ## Build the local Rust extension with maturin test: ## Run Python tests except recurring tests pytest $(PYTEST_ARGS) python/tests -integtest: ## Start LocalStack and run integration tests +integtest: ## Start S3-compatible services and run integration tests @if [ "$(KEEP_COMPOSE)" = "1" ]; then \ - docker compose -f $(COMPOSE_FILE) up -d --wait && \ - pytest --run-integration $(PYTEST_ARGS) python/tests/test_s3_ddb.py python/tests/test_namespace_integration.py; \ + docker compose -f $(COMPOSE_FILE) --profile $(COMPOSE_PROFILE) up -d --wait && \ + LANCE_COMMIT_CONFORMANCE_TRACE_DIR=$(COMMIT_CONFORMANCE_TRACE_DIR) pytest --run-integration $(PYTEST_ARGS) python/tests/test_s3_ddb.py python/tests/test_commit_conformance.py python/tests/test_namespace_integration.py; \ else \ - trap 'docker compose -f $(COMPOSE_FILE) down -v --remove-orphans --timeout 0 >/dev/null 2>&1 || true' EXIT; \ - docker compose -f $(COMPOSE_FILE) up -d --wait && \ - pytest --run-integration $(PYTEST_ARGS) python/tests/test_s3_ddb.py python/tests/test_namespace_integration.py; \ + trap 'docker compose -f $(COMPOSE_FILE) --profile $(COMPOSE_PROFILE) down -v --remove-orphans --timeout 0 >/dev/null 2>&1 || true' EXIT; \ + docker compose -f $(COMPOSE_FILE) --profile $(COMPOSE_PROFILE) up -d --wait && \ + LANCE_COMMIT_CONFORMANCE_TRACE_DIR=$(COMMIT_CONFORMANCE_TRACE_DIR) pytest --run-integration $(PYTEST_ARGS) python/tests/test_s3_ddb.py python/tests/test_commit_conformance.py python/tests/test_namespace_integration.py; \ fi doctest: ## Run Python doctests diff --git a/python/pyproject.toml b/python/pyproject.toml index d89d3671dca..e6ab41de2c0 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -115,7 +115,11 @@ include = [ "python/lance/util.py", "python/lance/arrow.py", "python/tests/test_arrow.py", + "python/tests/_commit_conformance.py", + "python/tests/_commit_fault_proxy.py", + "python/tests/test_commit_conformance.py", ] +extraPaths = ["python/tests"] # Dependencies like pyarrow make this difficult to enforce strictly. reportMissingTypeStubs = "warning" reportImportCycles = "error" @@ -132,6 +136,7 @@ markers = [ "slow", "torch: tests which rely on pytorch being installed", "recurring: marks tests as recurring tests", + "real_s3: tests that require the fail-closed real-AWS S3 runner", ] filterwarnings = [ 'error::FutureWarning', diff --git a/python/python/tests/_commit_conformance.py b/python/python/tests/_commit_conformance.py new file mode 100644 index 00000000000..4998a73209f --- /dev/null +++ b/python/python/tests/_commit_conformance.py @@ -0,0 +1,758 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The Lance Authors +"""Public-API drivers and final-state checks for commit conformance tests.""" + +from __future__ import annotations + +import json +import os +import signal +import subprocess +import sys +from dataclasses import dataclass +from typing import TYPE_CHECKING, Literal, Mapping, TypedDict, cast + +if TYPE_CHECKING: + from pathlib import Path + +CommitBackend = Literal["conditional", "dynamodb"] +CommitOperation = Literal["append", "compaction", "add_columns", "create_index"] +MaintenanceOperation = Literal["setup", "cleanup"] +WriterVariant = Literal["main", "a", "b"] + +BASE_ROW_COUNT = 200 +APPEND_ROWS = { + "main": list(range(200, 220)), + "a": list(range(200, 220)), + "b": list(range(300, 320)), +} +ADDED_COLUMNS = { + "main": ("double_id", "id * 2"), + "a": ("writer_a", "id + 1000"), + "b": ("writer_b", "id + 2000"), +} +INDEX_NAMES = { + "main": "id_idx", + "a": "id_idx_a", + "b": "id_idx_b", +} + + +@dataclass(frozen=True) +class BackendContract: + uri_scheme: str + commit_request_operation: str + requires_dynamodb_table: bool = False + + +@dataclass(frozen=True) +class OperationContract: + commit_request_occurrence: int = 1 + max_before_versions: int = 1 + + +BACKEND_CONTRACTS: Mapping[CommitBackend, BackendContract] = { + "conditional": BackendContract( + uri_scheme="s3", commit_request_operation="s3_manifest_create" + ), + "dynamodb": BackendContract( + uri_scheme="s3+ddb", + commit_request_operation="ddb_commit_create", + requires_dynamodb_table=True, + ), +} + +OPERATION_CONTRACTS: Mapping[CommitOperation, OperationContract] = { + "append": OperationContract(), + "compaction": OperationContract(commit_request_occurrence=2, max_before_versions=2), + "add_columns": OperationContract(), + "create_index": OperationContract(), +} + +TRANSACTION_OPERATIONS: Mapping[CommitOperation, str] = { + "append": "Append", + "compaction": "Rewrite", + "add_columns": "Merge", + "create_index": "CreateIndex", +} + + +class DatasetState(TypedDict): + rows: list[dict[str, int]] + schema: list[tuple[str, str]] + fragments: int + indices: list[dict[str, object]] + index_queries: dict[str, list[int]] + + +class TransactionState(TypedDict): + uuid: str + operation: str + + +class DatasetHealth(TypedDict): + versions: list[int] + opened_version: int + states: dict[str, DatasetState] + transactions: dict[str, TransactionState] + + +@dataclass(frozen=True) +class OperationResult: + status: Literal["success", "error", "killed", "crashed"] + returncode: int + error_type: str | None + error_message: str | None + committed_version: int | None + transaction_uuid: str | None + stdout: str + stderr: str + + +_OPERATION_SCRIPT = r""" +import json +import os +import sys +import time +from datetime import timedelta +from pathlib import Path + +import lance +import pyarrow as pa + +uri = os.environ["LANCE_CONFORMANCE_URI"] +storage_options = json.loads(os.environ["LANCE_CONFORMANCE_STORAGE_OPTIONS"]) +operation = os.environ["LANCE_CONFORMANCE_OPERATION"] +variant = os.environ.get("LANCE_CONFORMANCE_VARIANT", "main") +gate = os.environ.get("LANCE_CONFORMANCE_START_GATE") +if gate: + deadline = time.monotonic() + 30 + while not Path(gate).exists(): + if time.monotonic() >= deadline: + raise TimeoutError(f"start gate was not released: {gate}") + time.sleep(0.01) + +try: + result_dataset = None + if operation == "setup": + result_dataset = lance.write_dataset( + pa.table({"id": pa.array(range(200), type=pa.int64())}), + uri, + max_rows_per_file=100, + storage_options=storage_options, + ) + elif operation == "append": + ranges = { + "main": range(200, 220), + "a": range(200, 220), + "b": range(300, 320), + } + result_dataset = lance.write_dataset( + pa.table({"id": pa.array(ranges[variant], type=pa.int64())}), + uri, + mode="append", + storage_options=storage_options, + ) + elif operation == "cleanup": + dataset = lance.dataset(uri, storage_options=storage_options) + dataset.cleanup_old_versions( + older_than=timedelta(0), + retain_versions=100, + delete_unverified=True, + ) + else: + dataset = lance.dataset(uri, storage_options=storage_options) + if operation == "compaction": + dataset.optimize.compact_files( + target_rows_per_fragment=1000, + materialize_deletions=False, + num_threads=1, + ) + elif operation == "add_columns": + columns = { + "main": ("double_id", "id * 2"), + "a": ("writer_a", "id + 1000"), + "b": ("writer_b", "id + 2000"), + } + name, expression = columns[variant] + dataset.add_columns({name: expression}) + elif operation == "create_index": + names = {"main": "id_idx", "a": "id_idx_a", "b": "id_idx_b"} + dataset.create_scalar_index("id", "BTREE", name=names[variant]) + else: + raise ValueError(f"unknown operation: {operation}") + result_dataset = dataset + + result = {"status": "success"} + if result_dataset is not None: + committed_version = result_dataset.version + transaction = result_dataset.read_transaction(committed_version) + if transaction is None: + raise AssertionError( + "successful operation has no transaction at version " + f"{committed_version}" + ) + result.update({ + "committed_version": committed_version, + "transaction_uuid": transaction.uuid, + }) + print(json.dumps(result)) +except Exception as error: + print(json.dumps({ + "status": "error", + "error_type": type(error).__name__, + "error_message": str(error), + })) +""" + + +_HEALTH_SCRIPT = r""" +import json +import os + +import lance + +uri = os.environ["LANCE_CONFORMANCE_URI"] +storage_options = json.loads(os.environ["LANCE_CONFORMANCE_STORAGE_OPTIONS"]) +dataset = lance.dataset(uri, storage_options=storage_options) +versions = sorted(item["version"] for item in dataset.versions()) +states = {} +transactions = {} +for version in versions: + snapshot = dataset.checkout_version(version) + snapshot.validate() + transaction = snapshot.read_transaction(version) + if transaction is None: + raise AssertionError(f"version {version} has no readable transaction") + transactions[str(version)] = { + "uuid": transaction.uuid, + "operation": type(transaction.operation).__name__, + } + table = snapshot.to_table() + rows = sorted(table.to_pylist(), key=lambda row: row["id"]) + indices = [] + index_queries = {} + btree_id_indices = [] + for index in snapshot.list_indices(): + normalized = { + "name": index["name"], + "type": index["type"], + "fields": sorted(index["fields"]), + } + indices.append(normalized) + index_stats = snapshot.stats.index_stats(index["name"]) + if not index_stats: + raise AssertionError(f"index {index['name']} has no readable statistics") + if index["type"] == "BTree" and index["fields"] == ["id"]: + btree_id_indices.append(index["name"]) + if btree_id_indices: + scanner = snapshot.scanner(filter="id >= 190", use_scalar_index=True) + plan = scanner.explain_plan() + if "ScalarIndexQuery" not in plan or not any( + name in plan for name in btree_id_indices + ): + raise AssertionError( + f"no BTree id index was used by the health query: {plan}" + ) + query_rows = sorted(scanner.to_table()["id"].to_pylist()) + index_queries.update({name: query_rows for name in btree_id_indices}) + states[str(version)] = { + "rows": rows, + "schema": [[field.name, str(field.type)] for field in table.schema], + "fragments": len(snapshot.get_fragments()), + "indices": sorted(indices, key=lambda index: index["name"]), + "index_queries": index_queries, + } + +print(json.dumps({ + "versions": versions, + "opened_version": dataset.version, + "states": states, + "transactions": transactions, +}, sort_keys=True)) +""" + + +def dataset_uri( + backend: CommitBackend, + bucket: str, + key: str, + *, + dynamodb_table: str | None, +) -> str: + contract = BACKEND_CONTRACTS[backend] + if contract.requires_dynamodb_table and dynamodb_table is None: + raise ValueError("dynamodb_table is required for the dynamodb backend") + query = ( + f"?ddbTableName={dynamodb_table}" if contract.requires_dynamodb_table else "" + ) + return f"{contract.uri_scheme}://{bucket}/{key}{query}" + + +def commit_request_operation(backend: CommitBackend) -> str: + return BACKEND_CONTRACTS[backend].commit_request_operation + + +def commit_request_occurrence(operation: CommitOperation) -> int: + return OPERATION_CONTRACTS[operation].commit_request_occurrence + + +def operation_environment( + uri: str, + storage_options: dict[str, str], + operation: CommitOperation | MaintenanceOperation, + variant: WriterVariant = "main", + *, + start_gate: Path | None = None, +) -> dict[str, str]: + env = os.environ.copy() + env["LANCE_CONFORMANCE_URI"] = uri + env["LANCE_CONFORMANCE_STORAGE_OPTIONS"] = json.dumps(storage_options) + env["LANCE_CONFORMANCE_OPERATION"] = operation + env["LANCE_CONFORMANCE_VARIANT"] = variant + if start_gate is not None: + env["LANCE_CONFORMANCE_START_GATE"] = str(start_gate) + return env + + +def start_operation( + uri: str, + storage_options: dict[str, str], + operation: CommitOperation | MaintenanceOperation, + variant: WriterVariant = "main", + *, + start_gate: Path | None = None, +) -> subprocess.Popen[str]: + return subprocess.Popen( + [sys.executable, "-c", _OPERATION_SCRIPT], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=operation_environment( + uri, + storage_options, + operation, + variant, + start_gate=start_gate, + ), + text=True, + ) + + +def finish_operation( + process: subprocess.Popen[str], *, timeout: float = 180 +) -> OperationResult: + try: + stdout, stderr = process.communicate(timeout=timeout) + except subprocess.TimeoutExpired as error: + process.kill() + stdout, stderr = process.communicate() + raise AssertionError( + f"operation timed out after {timeout} seconds and was reaped\n" + f"stdout:\n{stdout}\nstderr:\n{stderr}" + ) from error + + if process.returncode != 0: + returncode = cast("int", process.returncode) + return OperationResult( + status="killed" if returncode == -signal.SIGKILL else "crashed", + returncode=returncode, + error_type=None, + error_message=None, + committed_version=None, + transaction_uuid=None, + stdout=stdout, + stderr=stderr, + ) + lines = [line for line in stdout.splitlines() if line.strip()] + if not lines: + raise AssertionError(f"operation produced no result\nstderr:\n{stderr}") + payload = json.loads(lines[-1]) + return OperationResult( + status=payload["status"], + returncode=0, + error_type=payload.get("error_type"), + error_message=payload.get("error_message"), + committed_version=payload.get("committed_version"), + transaction_uuid=payload.get("transaction_uuid"), + stdout=stdout, + stderr=stderr, + ) + + +def stop_operation(process: subprocess.Popen[str]) -> None: + """Kill and reap a child that a failed assertion would otherwise leak.""" + + if process.poll() is None: + process.kill() + process.communicate() + + +def execute_operation( + uri: str, + storage_options: dict[str, str], + operation: CommitOperation | MaintenanceOperation, + variant: WriterVariant = "main", + *, + timeout: float = 180, +) -> OperationResult: + return finish_operation( + start_operation(uri, storage_options, operation, variant), timeout=timeout + ) + + +def check_dataset_in_fresh_process( + uri: str, storage_options: dict[str, str], *, timeout: float = 180 +) -> DatasetHealth: + env = os.environ.copy() + env["LANCE_CONFORMANCE_URI"] = uri + env["LANCE_CONFORMANCE_STORAGE_OPTIONS"] = json.dumps(storage_options) + result = subprocess.run( + [sys.executable, "-c", _HEALTH_SCRIPT], + check=False, + capture_output=True, + env=env, + text=True, + timeout=timeout, + ) + assert result.returncode == 0, ( + "fresh-process dataset validation failed\n" + f"stdout:\n{result.stdout}\n" + f"stderr:\n{result.stderr}" + ) + lines = [line for line in result.stdout.splitlines() if line.strip()] + assert lines, f"fresh-process health check produced no output: {result.stderr}" + return cast("DatasetHealth", json.loads(lines[-1])) + + +def base_state() -> DatasetState: + return { + "rows": [{"id": value} for value in range(BASE_ROW_COUNT)], + "schema": [("id", "int64")], + "fragments": 2, + "indices": [], + "index_queries": {}, + } + + +def state_after_operation( + operation: CommitOperation, variant: WriterVariant = "main" +) -> DatasetState: + state = base_state() + if operation == "append": + state["rows"] = state["rows"] + [ + {"id": value} for value in APPEND_ROWS[variant] + ] + state["fragments"] = 3 + elif operation == "compaction": + state["fragments"] = 1 + elif operation == "add_columns": + name, _ = ADDED_COLUMNS[variant] + offset = {"main": 0, "a": 1000, "b": 2000}[variant] + multiplier = 2 if variant == "main" else 1 + state["rows"] = [ + {"id": value, name: value * multiplier + offset} + for value in range(BASE_ROW_COUNT) + ] + state["schema"] = [("id", "int64"), (name, "int64")] + elif operation == "create_index": + name = INDEX_NAMES[variant] + state["indices"] = [{"name": name, "type": "BTree", "fields": ["id"]}] + state["index_queries"] = {name: list(range(190, BASE_ROW_COUNT))} + return state + + +def combined_foreign_writer_states(operation: CommitOperation) -> list[DatasetState]: + states = [base_state(), state_after_operation(operation, "a")] + state_b = state_after_operation(operation, "b") + if state_b not in states: + states.append(state_b) + + if operation == "append": + both = base_state() + both["rows"] = both["rows"] + [ + {"id": value} for value in APPEND_ROWS["a"] + APPEND_ROWS["b"] + ] + both["fragments"] = 4 + states.append(both) + elif operation == "add_columns": + both = base_state() + both["rows"] = [ + {"id": value, "writer_a": value + 1000, "writer_b": value + 2000} + for value in range(BASE_ROW_COUNT) + ] + both["schema"] = [ + ("id", "int64"), + ("writer_a", "int64"), + ("writer_b", "int64"), + ] + states.append(both) + reversed_schema = cast("DatasetState", json.loads(json.dumps(both))) + reversed_schema["schema"] = [ + ("id", "int64"), + ("writer_b", "int64"), + ("writer_a", "int64"), + ] + states.append(reversed_schema) + elif operation == "create_index": + both = base_state() + both["indices"] = [ + {"name": INDEX_NAMES["a"], "type": "BTree", "fields": ["id"]}, + {"name": INDEX_NAMES["b"], "type": "BTree", "fields": ["id"]}, + ] + both["index_queries"] = { + INDEX_NAMES["a"]: list(range(190, BASE_ROW_COUNT)), + INDEX_NAMES["b"]: list(range(190, BASE_ROW_COUNT)), + } + states.append(both) + return states + + +def state_signature(state: DatasetState) -> str: + return json.dumps(state, sort_keys=True, separators=(",", ":")) + + +@dataclass(frozen=True) +class HistoryNode: + name: str + state: DatasetState + max_versions: int = 1 + + +@dataclass(frozen=True) +class HistoryContract: + nodes: tuple[HistoryNode, ...] + transitions: frozenset[tuple[str, str]] + latest_nodes: frozenset[str] + + +def operation_history_contract( + operation: CommitOperation, + *, + latest: Literal["before", "after", "either"], + attempts: int = 1, +) -> HistoryContract: + if attempts < 1: + raise ValueError(f"attempts must be at least 1, got {attempts}") + operation_contract = OPERATION_CONTRACTS[operation] + latest_nodes = { + "before": frozenset({"before"}), + "after": frozenset({"after"}), + "either": frozenset({"before", "after"}), + }[latest] + return HistoryContract( + nodes=( + HistoryNode( + "before", + base_state(), + max_versions=( + 1 + attempts * (operation_contract.max_before_versions - 1) + ), + ), + HistoryNode("after", state_after_operation(operation)), + ), + transitions=frozenset({("before", "after")}), + latest_nodes=latest_nodes, + ) + + +def foreign_writer_history_contract(operation: CommitOperation) -> HistoryContract: + states = combined_foreign_writer_states(operation) + if operation == "compaction": + # Each writer may reserve fragment IDs before one rewrite wins. A losing + # writer can commit its reservation after the winning rewrite without + # applying a second rewrite. + return HistoryContract( + nodes=( + HistoryNode("before", states[0], max_versions=3), + HistoryNode("after", states[1], max_versions=2), + ), + transitions=frozenset({("before", "after")}), + latest_nodes=frozenset({"after"}), + ) + + nodes = [HistoryNode("before", states[0])] + nodes.extend( + HistoryNode(f"effect_{index}", state) + for index, state in enumerate(states[1:], start=1) + ) + combined_start = 3 + transitions = { + ("before", "effect_1"), + ("before", "effect_2"), + } + transitions.update( + (single, f"effect_{index}") + for single in ("effect_1", "effect_2") + for index in range(combined_start, len(states)) + ) + return HistoryContract( + nodes=tuple(nodes), + transitions=frozenset(transitions), + latest_nodes=frozenset(node.name for node in nodes[1:]), + ) + + +def recover_unknown_operation( + uri: str, + storage_options: dict[str, str], + operation: CommitOperation, +) -> tuple[Literal["already_applied", "retried"], DatasetHealth]: + before = base_state() + after = state_after_operation(operation) + health = check_dataset_in_fresh_process(uri, storage_options) + latest = health["states"][str(health["versions"][-1])] + if state_signature(latest) == state_signature(after): + return "already_applied", health + if state_signature(latest) != state_signature(before): + raise AssertionError( + f"cannot recover operation from a partial or foreign state: {latest}" + ) + + retry = execute_operation(uri, storage_options, operation) + assert retry.status == "success", retry + return "retried", check_dataset_in_fresh_process(uri, storage_options) + + +def assert_history_conforms( + health: DatasetHealth, + contract: HistoryContract, +) -> None: + versions = health["versions"] + assert versions, "dataset has no visible versions" + assert versions == list(range(1, versions[-1] + 1)), ( + f"visible versions are not contiguous: {versions}" + ) + assert health["opened_version"] == versions[-1], ( + "opening the dataset did not resolve to the latest visible version: " + f"opened={health['opened_version']}, visible={versions}" + ) + + transaction_versions = {int(version) for version in health["transactions"]} + assert transaction_versions == set(versions), ( + "transaction metadata does not cover every visible version: " + f"transactions={sorted(transaction_versions)}, visible={versions}" + ) + transaction_uuids = [ + health["transactions"][str(version)]["uuid"] for version in versions + ] + assert all(transaction_uuids), "every visible version must have a transaction UUID" + assert len(transaction_uuids) == len(set(transaction_uuids)), ( + f"a transaction UUID was committed more than once: {transaction_uuids}" + ) + + nodes_by_signature = {state_signature(node.state): node for node in contract.nodes} + assert len(nodes_by_signature) == len(contract.nodes), ( + "history contract contains indistinguishable state nodes" + ) + observed_nodes: list[str] = [] + node_counts: dict[str, int] = {} + for version in versions: + state = health["states"][str(version)] + signature = state_signature(state) + node = nodes_by_signature.get(signature) + assert node is not None, ( + f"version {version} is not a complete contract state: {state}" + ) + observed_nodes.append(node.name) + node_counts[node.name] = node_counts.get(node.name, 0) + 1 + assert node_counts[node.name] <= node.max_versions, ( + f"state {node.name!r} appears in too many versions: {observed_nodes}" + ) + ids = [row["id"] for row in state["rows"]] + assert len(ids) == len(set(ids)), ( + f"version {version} contains duplicate row IDs" + ) + + assert observed_nodes[0] == "before", ( + f"history must start from the setup state: {observed_nodes}" + ) + for previous, current in zip(observed_nodes, observed_nodes[1:]): + if previous != current: + assert (previous, current) in contract.transitions, ( + f"illegal history transition {previous!r} -> {current!r}: " + f"{observed_nodes}" + ) + assert observed_nodes[-1] in contract.latest_nodes, ( + f"latest state {observed_nodes[-1]!r} is not allowed: {observed_nodes}" + ) + + +def assert_successful_result_is_visible( + result: OperationResult, + health: DatasetHealth, + operation: CommitOperation, + variant: WriterVariant = "main", +) -> None: + if result.status != "success": + return + assert result.committed_version is not None, result + assert result.transaction_uuid is not None, result + assert result.committed_version in health["versions"], ( + f"successful operation returned invisible version {result.committed_version}: " + f"{health['versions']}" + ) + transaction = health["transactions"][str(result.committed_version)] + assert transaction["uuid"] == result.transaction_uuid, ( + "successful operation does not own its reported visible version: " + f"result={result}, transaction={transaction}" + ) + expected_transaction_operation = TRANSACTION_OPERATIONS[operation] + assert transaction["operation"] == expected_transaction_operation, ( + "successful operation reported a version committed by a different intent: " + f"expected={expected_transaction_operation}, transaction={transaction}" + ) + + state = health["states"][str(result.committed_version)] + if operation == "append": + visible_ids = {row["id"] for row in state["rows"]} + assert set(APPEND_ROWS[variant]).issubset(visible_ids), ( + f"successful append effect is absent at version {result.committed_version}" + ) + elif operation == "compaction": + assert state["fragments"] == 1, ( + "successful compaction effect is absent at version " + f"{result.committed_version}" + ) + elif operation == "add_columns": + column, _ = ADDED_COLUMNS[variant] + assert column in {name for name, _ in state["schema"]}, ( + "successful add-columns effect is absent at version " + f"{result.committed_version}" + ) + elif operation == "create_index": + index_name = INDEX_NAMES[variant] + assert index_name in {index["name"] for index in state["indices"]}, ( + "successful create-index effect is absent at version " + f"{result.committed_version}" + ) + + +def assert_results_match_visible_intents( + results: tuple[OperationResult, ...], + health: DatasetHealth, + operation: CommitOperation, +) -> None: + """Match every unambiguous writer result to exactly one visible intent.""" + successful_results = [result for result in results if result.status == "success"] + returned_transactions: set[tuple[int, str]] = set() + for result in successful_results: + assert result.committed_version is not None, result + assert result.transaction_uuid is not None, result + returned_transactions.add((result.committed_version, result.transaction_uuid)) + assert len(returned_transactions) == len(successful_results), ( + "distinct successful writers reported the same transaction: " + f"{successful_results}" + ) + + expected_operation = TRANSACTION_OPERATIONS[operation] + visible_transactions = { + (version, transaction["uuid"]) + for version in health["versions"] + if (transaction := health["transactions"][str(version)])["operation"] + == expected_operation + } + assert returned_transactions == visible_transactions, ( + "unambiguous writer results do not match visible commit intents: " + f"returned={sorted(returned_transactions)}, " + f"visible={sorted(visible_transactions)}" + ) diff --git a/python/python/tests/_commit_fault_proxy.py b/python/python/tests/_commit_fault_proxy.py new file mode 100644 index 00000000000..9b05f9dfc70 --- /dev/null +++ b/python/python/tests/_commit_fault_proxy.py @@ -0,0 +1,719 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The Lance Authors +"""Deterministic storage fault injection for commit conformance tests.""" + +from __future__ import annotations + +import hashlib +import http.client +import json +import os +import socket +import ssl +import threading +from dataclasses import asdict, dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import TYPE_CHECKING, Literal, cast +from urllib.parse import urlsplit + +if TYPE_CHECKING: + from email.message import Message + from pathlib import Path + + +@dataclass(frozen=True) +class RequestTrace: + """One request as observed on both sides of the proxy.""" + + sequence: int + method: str + path: str + operation: str + identity: str + resource: str + request_sha256: str + forwarded: bool + upstream_status: int | None + client_status: int | None + injected: bool + fault_phase: str | None + + +@dataclass(frozen=True) +class FaultPlan: + """A serializable deterministic fault at a classified request boundary.""" + + operation: str + occurrence: int = 1 + phase: Literal["before", "after"] = "after" + response_losses: int = 1 + verification_response_losses: int = 0 + kill_pid: int | None = None + + def __post_init__(self) -> None: + if self.occurrence < 1: + raise ValueError(f"occurrence must be at least 1, got {self.occurrence}") + if self.response_losses < 1: + raise ValueError( + f"response_losses must be at least 1, got {self.response_losses}" + ) + if self.verification_response_losses < 0: + raise ValueError( + "verification_response_losses must be non-negative, got " + f"{self.verification_response_losses}" + ) + + +@dataclass(frozen=True) +class _ClassifiedRequest: + method: str + path: str + operation: str + identity: str + resource: str + request_sha256: str + mutation: bool + + +@dataclass +class _ArmedFault: + plan: FaultPlan + matches: int = 0 + response_losses_remaining: int = 0 + target_identity: str | None = None + target_resource: str | None = None + + def __post_init__(self) -> None: + self.response_losses_remaining = self.plan.response_losses + + +@dataclass +class _VerificationFault: + operation: str + identity: str + remaining: int + + +@dataclass +class _BarrierFault: + operation: str + parties: int + barrier: threading.Barrier + identity: str | None = None + arrivals: int = 0 + + +@dataclass(frozen=True) +class _ResponseAction: + kind: Literal["none", "replace", "kill"] + phase: str | None = None + pid: int | None = None + + +class _ProxyState: + def __init__(self) -> None: + self._lock = threading.Lock() + self._fault: _ArmedFault | None = None + self._verification_fault: _VerificationFault | None = None + self._barrier_fault: _BarrierFault | None = None + self._traces: list[RequestTrace] = [] + self._next_sequence = 1 + + def arm(self, plan: FaultPlan) -> None: + with self._lock: + self._fault = _ArmedFault(plan) + self._verification_fault = None + + def arm_barrier(self, *, operation: str, parties: int) -> None: + if parties < 2: + raise ValueError(f"parties must be at least 2, got {parties}") + with self._lock: + self._barrier_fault = _BarrierFault( + operation=operation, + parties=parties, + barrier=threading.Barrier(parties, timeout=30), + ) + + def clear_faults(self) -> None: + with self._lock: + self._fault = None + self._verification_fault = None + self._barrier_fault = None + + def clear_traces(self) -> None: + with self._lock: + self._traces.clear() + self._next_sequence = 1 + + def wait_at_barrier(self, request: _ClassifiedRequest) -> None: + barrier: threading.Barrier | None = None + with self._lock: + fault = self._barrier_fault + if fault is None or not self._matches_operation(fault.operation, request): + return + if fault.identity is None: + fault.identity = request.identity + if request.identity != fault.identity or fault.arrivals >= fault.parties: + return + fault.arrivals += 1 + barrier = fault.barrier + assert barrier is not None + barrier.wait() + + def before_request(self, request: _ClassifiedRequest) -> _ResponseAction: + with self._lock: + fault = self._fault + if fault is None or fault.plan.phase != "before": + return _ResponseAction("none") + if not self._fault_targets_request(fault, request): + return _ResponseAction("none") + fault.response_losses_remaining -= 1 + if fault.response_losses_remaining == 0: + self._fault = None + return _ResponseAction("replace", "before") + + def after_response( + self, request: _ClassifiedRequest, upstream_status: int + ) -> _ResponseAction: + if not 200 <= upstream_status < 300: + return _ResponseAction("none") + + with self._lock: + verification = self._verification_fault + if ( + verification is not None + and request.operation == verification.operation + and request.identity == verification.identity + ): + verification.remaining -= 1 + if verification.remaining == 0: + self._verification_fault = None + return _ResponseAction("replace", "verification") + + fault = self._fault + if fault is None or fault.plan.phase != "after": + return _ResponseAction("none") + if not self._fault_targets_request(fault, request): + return _ResponseAction("none") + + if fault.plan.verification_response_losses > 0: + verification_operation = { + "s3_manifest_create": "s3_manifest_read", + "ddb_commit_create": "ddb_commit_read", + }.get(request.operation) + if verification_operation is not None: + self._verification_fault = _VerificationFault( + operation=verification_operation, + identity=request.identity, + remaining=fault.plan.verification_response_losses, + ) + + fault.response_losses_remaining -= 1 + if fault.response_losses_remaining == 0: + self._fault = None + if fault.plan.kill_pid is not None: + return _ResponseAction("kill", "after", fault.plan.kill_pid) + return _ResponseAction("replace", "after") + + def _fault_targets_request( + self, fault: _ArmedFault, request: _ClassifiedRequest + ) -> bool: + if fault.target_identity is not None: + return ( + request.identity == fault.target_identity + and request.resource == fault.target_resource + ) + if not self._matches_operation(fault.plan.operation, request): + return False + fault.matches += 1 + if fault.matches != fault.plan.occurrence: + return False + fault.target_identity = request.identity + fault.target_resource = request.resource + return True + + @staticmethod + def _matches_operation(operation: str, request: _ClassifiedRequest) -> bool: + return operation == request.operation or ( + operation == "mutation" and request.mutation + ) + + def record( + self, + request: _ClassifiedRequest, + *, + forwarded: bool, + upstream_status: int | None, + client_status: int | None, + injected: bool, + fault_phase: str | None, + ) -> None: + with self._lock: + sequence = self._next_sequence + self._next_sequence += 1 + self._traces.append( + RequestTrace( + sequence=sequence, + method=request.method, + path=request.path, + operation=request.operation, + identity=request.identity, + resource=request.resource, + request_sha256=request.request_sha256, + forwarded=forwarded, + upstream_status=upstream_status, + client_status=client_status, + injected=injected, + fault_phase=fault_phase, + ) + ) + + def traces(self) -> list[RequestTrace]: + with self._lock: + return list(self._traces) + + +class _ProxyServer(ThreadingHTTPServer): + daemon_threads = True + + def __init__( + self, + server_address: tuple[str, int], + upstream_scheme: str, + upstream_host: str, + upstream_port: int, + upstream_connect_host: str | None, + state: _ProxyState, + ) -> None: + super().__init__(server_address, _CommitProxyHandler) + self.upstream_scheme = upstream_scheme + self.upstream_host = upstream_host + self.upstream_port = upstream_port + self.upstream_connect_host = upstream_connect_host + self.state = state + + +class _FixedAddressHTTPSConnection(http.client.HTTPSConnection): + """Connect to a fixed address while authenticating the original TLS host.""" + + def __init__( + self, + host: str, + port: int, + *, + connect_host: str, + timeout: float, + ) -> None: + self._tls_context = ssl.create_default_context() + super().__init__(host, port, timeout=timeout, context=self._tls_context) + self._fixed_connect_host = connect_host + + def connect(self) -> None: + self.sock = socket.create_connection( + (self._fixed_connect_host, self.port), + self.timeout, + ) + self.sock = self._tls_context.wrap_socket(self.sock, server_hostname=self.host) + + +class _CommitProxyHandler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_DELETE(self) -> None: # noqa: N802 + self._proxy_request() + + def do_GET(self) -> None: # noqa: N802 + self._proxy_request() + + def do_HEAD(self) -> None: # noqa: N802 + self._proxy_request() + + def do_PATCH(self) -> None: # noqa: N802 + self._proxy_request() + + def do_POST(self) -> None: # noqa: N802 + self._proxy_request() + + def do_PUT(self) -> None: # noqa: N802 + self._proxy_request() + + def log_message(self, format: str, *args: object) -> None: + pass + + def _proxy_request(self) -> None: + server = cast("_ProxyServer", self.server) + body = self._read_request_body() + request = _classify_request(self.command, self.path, self.headers, body) + + before_action = server.state.before_request(request) + if before_action.kind == "replace": + self._send_injected_failure(request) + server.state.record( + request, + forwarded=False, + upstream_status=None, + client_status=503, + injected=True, + fault_phase=before_action.phase, + ) + return + + server.state.wait_at_barrier(request) + headers = { + key: value + for key, value in self.headers.items() + if key.lower() + not in {"connection", "content-length", "expect", "transfer-encoding"} + } + headers["Content-Length"] = str(len(body)) + + if server.upstream_scheme == "https": + if server.upstream_connect_host is None: + upstream: http.client.HTTPConnection = http.client.HTTPSConnection( + server.upstream_host, + server.upstream_port, + timeout=30, + ) + else: + upstream = _FixedAddressHTTPSConnection( + server.upstream_host, + server.upstream_port, + connect_host=server.upstream_connect_host, + timeout=30, + ) + else: + upstream = http.client.HTTPConnection( + server.upstream_connect_host or server.upstream_host, + server.upstream_port, + timeout=30, + ) + try: + upstream.request(self.command, self.path, body=body, headers=headers) + response = upstream.getresponse() + response_body = response.read() + upstream_status = response.status + upstream_headers = response.getheaders() + finally: + upstream.close() + + action = server.state.after_response(request, upstream_status) + if action.kind == "kill": + server.state.record( + request, + forwarded=True, + upstream_status=upstream_status, + client_status=None, + injected=True, + fault_phase=action.phase, + ) + self.close_connection = True + assert action.pid is not None + os.kill(action.pid, 9) + return + + if action.kind == "replace": + client_status = 503 + self._send_injected_failure(request) + else: + client_status = upstream_status + self._send_upstream_response( + status=upstream_status, + headers=upstream_headers, + body=response_body, + ) + + server.state.record( + request, + forwarded=True, + upstream_status=upstream_status, + client_status=client_status, + injected=action.kind != "none", + fault_phase=action.phase, + ) + + def _read_request_body(self) -> bytes: + content_length = self.headers.get("Content-Length") + if content_length is not None: + return self.rfile.read(int(content_length)) + if self.headers.get("Transfer-Encoding", "").lower() != "chunked": + return b"" + + body = bytearray() + while True: + size_line = self.rfile.readline() + if not size_line: + raise ConnectionError("unexpected EOF while reading chunk size") + size = int(size_line.split(b";", 1)[0].strip(), 16) + if size == 0: + while self.rfile.readline() not in {b"\r\n", b"\n", b""}: + pass + break + body.extend(self.rfile.read(size)) + if self.rfile.read(2) != b"\r\n": + raise ConnectionError("invalid chunk terminator") + return bytes(body) + + def _send_injected_failure(self, request: _ClassifiedRequest) -> None: + if request.operation.startswith("ddb_"): + body = json.dumps( + { + "__type": "com.amazonaws.dynamodb.v20120810#InternalServerError", + "message": "injected commit conformance failure", + } + ).encode() + content_type = "application/x-amz-json-1.0" + else: + body = ( + b'' + b"InternalError" + b"injected commit conformance failure" + b"lance-commit-conformance" + ) + content_type = "application/xml" + self.send_response(503) + self.send_header("Content-Type", content_type) + self.send_header( + "Content-Length", "0" if self.command == "HEAD" else str(len(body)) + ) + self.send_header("Connection", "close") + self.end_headers() + if self.command != "HEAD": + self.wfile.write(body) + self.close_connection = True + + def _send_upstream_response( + self, + *, + status: int, + headers: list[tuple[str, str]], + body: bytes, + ) -> None: + self.send_response(status) + for key, value in headers: + if key.lower() not in { + "connection", + "content-length", + "date", + "server", + "transfer-encoding", + }: + self.send_header(key, value) + if self.command == "HEAD": + content_length = next( + (value for key, value in headers if key.lower() == "content-length"), + "0", + ) + self.send_header("Content-Length", content_length) + else: + self.send_header("Content-Length", str(len(body))) + self.send_header("Connection", "close") + self.end_headers() + if self.command != "HEAD": + self.wfile.write(body) + self.close_connection = True + + +def _classify_request( + method: str, + path: str, + headers: Message[str, str], + body: bytes, +) -> _ClassifiedRequest: + target = headers.get("X-Amz-Target", "") + if target: + return _classify_dynamodb_request(method, path, target, body) + return _classify_s3_request(method, path, headers, body) + + +def _classify_dynamodb_request( + method: str, path: str, target: str, body: bytes +) -> _ClassifiedRequest: + operation_name = target.rsplit(".", 1)[-1] + try: + payload = json.loads(body or b"{}") + except json.JSONDecodeError: + payload = {} + item = payload.get("Item") or payload.get("Key") or {} + base_uri = item.get("base_uri", {}).get("S", "") + version = item.get("version", {}).get("N", "") + identity = f"{base_uri}@{version}" if base_uri or version else target + resource = item.get("path", {}).get("S", identity) + condition = payload.get("ConditionExpression", "") + + if operation_name == "PutItem" and "attribute_not_exists" in condition: + operation = "ddb_commit_create" + elif operation_name == "PutItem" and "attribute_exists" in condition: + operation = "ddb_commit_finalize" + elif operation_name == "PutItem": + operation = "ddb_put" + elif operation_name == "GetItem": + operation = "ddb_commit_read" + elif operation_name == "Query": + operation = "ddb_query" + elif operation_name == "DeleteItem": + operation = "ddb_delete" + else: + operation = f"ddb_{operation_name.lower()}" + + return _ClassifiedRequest( + method=method, + path=path, + operation=operation, + identity=identity, + resource=resource, + request_sha256=hashlib.sha256(body).hexdigest(), + mutation=operation + in { + "ddb_commit_create", + "ddb_commit_finalize", + "ddb_put", + "ddb_delete", + }, + ) + + +def _classify_s3_request( + method: str, + path: str, + headers: Message[str, str], + body: bytes, +) -> _ClassifiedRequest: + request_path = urlsplit(path).path + is_manifest = "/_versions/" in request_path and ".manifest" in request_path + if ( + method == "PUT" + and headers.get("If-None-Match") == "*" + and is_manifest + and request_path.endswith(".manifest") + ): + operation = "s3_manifest_create" + elif method in {"GET", "HEAD"} and is_manifest: + operation = "s3_manifest_read" + elif method == "DELETE": + operation = "s3_delete" + elif method == "PUT" and headers.get("X-Amz-Copy-Source"): + operation = "s3_copy" + elif method == "PUT": + operation = "s3_put" + elif method == "POST": + operation = "s3_post" + else: + operation = f"s3_{method.lower()}" + + return _ClassifiedRequest( + method=method, + path=path, + operation=operation, + identity=request_path, + resource=request_path, + request_sha256=hashlib.sha256(body).hexdigest(), + mutation=method in {"DELETE", "PATCH", "POST", "PUT"}, + ) + + +class CommitFaultProxy: + """Forward storage traffic while injecting replayable response faults.""" + + def __init__( + self, + upstream_endpoint: str, + *, + listen_port: int = 0, + upstream_connect_host: str | None = None, + ) -> None: + upstream = urlsplit(upstream_endpoint) + if upstream.scheme not in {"http", "https"} or upstream.hostname is None: + raise ValueError( + f"upstream_endpoint must be an HTTP(S) URL, got {upstream_endpoint}" + ) + self._state = _ProxyState() + self._server = _ProxyServer( + ("127.0.0.1", listen_port), + upstream.scheme, + upstream.hostname, + upstream.port or (443 if upstream.scheme == "https" else 80), + upstream_connect_host, + self._state, + ) + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + self._fault_plans: list[FaultPlan] = [] + + @property + def endpoint(self) -> str: + host, port = cast("tuple[str, int]", self._server.server_address) + return f"http://{host}:{port}" + + @property + def port(self) -> int: + _, port = cast("tuple[str, int]", self._server.server_address) + return port + + def arm(self, plan: FaultPlan) -> None: + self._fault_plans.append(plan) + self._state.arm(plan) + + def arm_barrier(self, *, operation: str, parties: int = 2) -> None: + self._state.arm_barrier(operation=operation, parties=parties) + + def clear_faults(self) -> None: + self._state.clear_faults() + + def clear_traces(self) -> None: + self._state.clear_traces() + self._fault_plans.clear() + + def traces(self) -> list[RequestTrace]: + return self._state.traces() + + def write_trace( + self, + path: Path, + *, + metadata: dict[str, object] | None = None, + ) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + payload = { + "metadata": metadata or {}, + "fault_plans": [ + { + **{ + key: value + for key, value in asdict(plan).items() + if key != "kill_pid" + }, + "kill_process": plan.kill_pid is not None, + } + for plan in self._fault_plans + ], + "requests": [asdict(trace) for trace in self.traces()], + } + path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + + def __enter__(self) -> CommitFaultProxy: + self._thread.start() + return self + + def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None: + self._server.shutdown() + self._server.server_close() + self._thread.join(timeout=5) + + +def read_fault_plans(path: Path, *, kill_pid: int | None = None) -> list[FaultPlan]: + """Load fault plans, rebinding any process-kill action explicitly.""" + + payload = json.loads(path.read_text()) + if any(plan.get("kill_process") for plan in payload.get("fault_plans", [])): + if kill_pid is None: + raise ValueError("replaying a process-kill fault requires a new kill_pid") + return [ + FaultPlan( + operation=str(raw["operation"]), + occurrence=int(raw["occurrence"]), + phase=cast("Literal['before', 'after']", raw["phase"]), + response_losses=int(raw["response_losses"]), + verification_response_losses=int(raw["verification_response_losses"]), + kill_pid=kill_pid if raw.get("kill_process") else None, + ) + for raw in payload.get("fault_plans", []) + ] diff --git a/python/python/tests/test_commit_conformance.py b/python/python/tests/test_commit_conformance.py new file mode 100644 index 00000000000..d011579d46e --- /dev/null +++ b/python/python/tests/test_commit_conformance.py @@ -0,0 +1,986 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The Lance Authors +"""Black-box commit protocol conformance tests over S3-compatible services.""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import uuid +from contextlib import contextmanager +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Iterator +from urllib.parse import urlsplit + +import boto3 +import pytest +from _commit_conformance import ( + CommitBackend, + CommitOperation, + DatasetHealth, + DatasetState, + OperationResult, + assert_history_conforms, + assert_results_match_visible_intents, + assert_successful_result_is_visible, + base_state, + check_dataset_in_fresh_process, + commit_request_occurrence, + commit_request_operation, + dataset_uri, + execute_operation, + finish_operation, + foreign_writer_history_contract, + operation_history_contract, + recover_unknown_operation, + start_operation, + state_after_operation, + stop_operation, +) +from _commit_fault_proxy import ( + CommitFaultProxy, + FaultPlan, + RequestTrace, + read_fault_plans, +) +from botocore.config import Config + +S3_ENDPOINT = os.environ.get("LANCE_CONFORMANCE_S3_ENDPOINT", "http://127.0.0.1:9000") +DYNAMODB_ENDPOINT = os.environ.get( + "LANCE_CONFORMANCE_DYNAMODB_ENDPOINT", "http://127.0.0.1:4566" +) +AWS_REGION = os.environ.get("AWS_DEFAULT_REGION", "us-east-1") +LOCAL_AWS_ACCESS_KEY_ID = os.environ.get( + "LANCE_CONFORMANCE_AWS_ACCESS_KEY_ID", "ACCESS_KEY" +) +LOCAL_AWS_SECRET_ACCESS_KEY = os.environ.get( + "LANCE_CONFORMANCE_AWS_SECRET_ACCESS_KEY", "SECRET_KEY" +) + +OPERATIONS: tuple[CommitOperation, ...] = ( + "append", + "compaction", + "add_columns", + "create_index", +) +BACKENDS: tuple[CommitBackend, ...] = ("conditional", "dynamodb") + + +@dataclass(frozen=True) +class ConformanceCase: + backend: CommitBackend + operation: CommitOperation + uri: str + storage_options: dict[str, str] + s3_proxy: CommitFaultProxy + dynamodb_proxy: CommitFaultProxy + + @property + def commit_proxy(self) -> CommitFaultProxy: + return self.s3_proxy if self.backend == "conditional" else self.dynamodb_proxy + + +def boto3_client(service: str, *, endpoint_url: str | None = None): + return boto3.client( + service, + endpoint_url=endpoint_url, + region_name=AWS_REGION, + aws_access_key_id=LOCAL_AWS_ACCESS_KEY_ID, + aws_secret_access_key=LOCAL_AWS_SECRET_ACCESS_KEY, + ) + + +def delete_bucket(s3, bucket: str) -> None: + paginator = s3.get_paginator("list_objects_v2") + for page in paginator.paginate(Bucket=bucket): + objects = [{"Key": item["Key"]} for item in page.get("Contents", [])] + if objects: + s3.delete_objects(Bucket=bucket, Delete={"Objects": objects}) + s3.delete_bucket(Bucket=bucket) + + +@pytest.fixture(scope="module") +def conformance_bucket() -> Iterator[str]: + s3 = boto3_client("s3", endpoint_url=S3_ENDPOINT) + bucket = f"lance-commit-conformance-{uuid.uuid4().hex[:16]}" + s3.create_bucket(Bucket=bucket) + yield bucket + delete_bucket(s3, bucket) + + +@pytest.fixture(scope="module") +def conformance_dynamodb_table() -> Iterator[str]: + dynamodb = boto3_client("dynamodb", endpoint_url=DYNAMODB_ENDPOINT) + table = f"lance-commit-conformance-{uuid.uuid4().hex[:16]}" + dynamodb.create_table( + TableName=table, + KeySchema=[ + {"AttributeName": "base_uri", "KeyType": "HASH"}, + {"AttributeName": "version", "KeyType": "RANGE"}, + ], + AttributeDefinitions=[ + {"AttributeName": "base_uri", "AttributeType": "S"}, + {"AttributeName": "version", "AttributeType": "N"}, + ], + BillingMode="PAY_PER_REQUEST", + ) + dynamodb.get_waiter("table_exists").wait(TableName=table) + yield table + dynamodb.delete_table(TableName=table) + + +def storage_options(*, s3_endpoint: str, dynamodb_endpoint: str) -> dict[str, str]: + return { + "allow_http": "true", + "aws_access_key_id": LOCAL_AWS_ACCESS_KEY_ID, + "aws_secret_access_key": LOCAL_AWS_SECRET_ACCESS_KEY, + "aws_region": AWS_REGION, + "aws_endpoint": s3_endpoint, + "aws_virtual_hosted_style_request": "false", + "client_max_retries": "2", + "client_retry_timeout": "5", + "dynamodb_endpoint": dynamodb_endpoint, + } + + +@contextmanager +def make_case( + *, + backend: CommitBackend, + operation: CommitOperation, + bucket: str, + dynamodb_table: str, +) -> Iterator[ConformanceCase]: + with ( + CommitFaultProxy(S3_ENDPOINT) as s3_proxy, + CommitFaultProxy(DYNAMODB_ENDPOINT) as dynamodb_proxy, + ): + options = storage_options( + s3_endpoint=s3_proxy.endpoint, + dynamodb_endpoint=dynamodb_proxy.endpoint, + ) + uri = dataset_uri( + backend, + bucket, + f"{uuid.uuid4().hex}.lance", + dynamodb_table=dynamodb_table, + ) + setup = execute_operation(uri, options, "setup") + assert setup.status == "success", setup + s3_proxy.clear_traces() + dynamodb_proxy.clear_traces() + yield ConformanceCase( + backend=backend, + operation=operation, + uri=uri, + storage_options=options, + s3_proxy=s3_proxy, + dynamodb_proxy=dynamodb_proxy, + ) + + +def trace_directory(tmp_path: Path) -> Path: + configured = os.environ.get("LANCE_COMMIT_CONFORMANCE_TRACE_DIR") + return Path(configured) if configured else tmp_path / "commit-conformance-traces" + + +def persist_case_traces( + case: ConformanceCase, + tmp_path: Path, + scenario: str, + *, + metadata: dict[str, object] | None = None, +) -> None: + root = trace_directory(tmp_path) + stem = f"{case.backend}-{case.operation}-{scenario}" + shared_metadata = { + "backend": case.backend, + "operation": case.operation, + "scenario": scenario, + **(metadata or {}), + } + case.s3_proxy.write_trace( + root / f"{stem}-s3.json", + metadata={**shared_metadata, "service": "s3"}, + ) + case.dynamodb_proxy.write_trace( + root / f"{stem}-dynamodb.json", + metadata={**shared_metadata, "service": "dynamodb"}, + ) + + +def assert_ambiguous_commit_trace( + case: ConformanceCase, *, lose_verification_reads: bool +) -> None: + operation = commit_request_operation(case.backend) + traces = case.commit_proxy.traces() + injected_commits = [ + trace + for trace in traces + if trace.operation == operation + and trace.injected + and trace.fault_phase == "after" + ] + assert len(injected_commits) == 1, traces + commit = injected_commits[0] + assert commit.forwarded + assert commit.upstream_status is not None + assert 200 <= commit.upstream_status < 300 + assert commit.client_status == 503 + assert any( + trace.identity == commit.identity + and trace.operation == operation + and trace.upstream_status in {400, 409, 412} + for trace in traces + ), traces + + verification_operation = ( + "s3_manifest_read" if case.backend == "conditional" else "ddb_commit_read" + ) + injected_verification = [ + trace + for trace in traces + if trace.identity == commit.identity + and trace.operation == verification_operation + and trace.injected + ] + assert bool(injected_verification) == lose_verification_reads + + +@pytest.mark.integration +@pytest.mark.parametrize("backend", BACKENDS) +@pytest.mark.parametrize("operation", OPERATIONS) +@pytest.mark.parametrize("lose_verification_reads", [False, True]) +def test_commit_response_loss_preserves_final_state( + conformance_bucket: str, + conformance_dynamodb_table: str, + tmp_path: Path, + backend: CommitBackend, + operation: CommitOperation, + lose_verification_reads: bool, +) -> None: + scenario = ( + "response-and-verification-lost" if lose_verification_reads else "response-lost" + ) + with make_case( + backend=backend, + operation=operation, + bucket=conformance_bucket, + dynamodb_table=conformance_dynamodb_table, + ) as case: + plan = FaultPlan( + operation=commit_request_operation(backend), + occurrence=commit_request_occurrence(operation), + verification_response_losses=100 if lose_verification_reads else 0, + ) + case.commit_proxy.arm(plan) + result: OperationResult | None = None + try: + result = execute_operation(case.uri, case.storage_options, operation) + assert_ambiguous_commit_trace( + case, lose_verification_reads=lose_verification_reads + ) + case.s3_proxy.clear_faults() + case.dynamodb_proxy.clear_faults() + health = check_dataset_in_fresh_process(case.uri, case.storage_options) + assert_history_conforms( + health, operation_history_contract(operation, latest="after") + ) + expected_status = "error" if lose_verification_reads else "success" + assert result.status == expected_status, result + assert_successful_result_is_visible(result, health, operation) + finally: + persist_case_traces( + case, + tmp_path, + scenario, + metadata={ + "operation_result": asdict(result) if result is not None else None, + }, + ) + + +@pytest.mark.integration +@pytest.mark.parametrize("backend", BACKENDS) +@pytest.mark.parametrize("operation", OPERATIONS) +def test_foreign_writer_cannot_corrupt_visible_versions( + conformance_bucket: str, + conformance_dynamodb_table: str, + tmp_path: Path, + backend: CommitBackend, + operation: CommitOperation, +) -> None: + with make_case( + backend=backend, + operation=operation, + bucket=conformance_bucket, + dynamodb_table=conformance_dynamodb_table, + ) as case: + gate = tmp_path / f"start-{backend}-{operation}" + case.commit_proxy.arm_barrier( + operation=commit_request_operation(backend), parties=2 + ) + writer_a = start_operation( + case.uri, + case.storage_options, + operation, + "a", + start_gate=gate, + ) + writer_b = start_operation( + case.uri, + case.storage_options, + operation, + "b", + start_gate=gate, + ) + result_a: OperationResult | None = None + result_b: OperationResult | None = None + try: + gate.touch() + result_a = finish_operation(writer_a) + result_b = finish_operation(writer_b) + traces = case.commit_proxy.traces() + operation_name = commit_request_operation(backend) + commit_traces = [ + trace for trace in traces if trace.operation == operation_name + ] + assert any( + trace.upstream_status is not None and 200 <= trace.upstream_status < 300 + for trace in commit_traces + ), commit_traces + assert any( + trace.upstream_status in {400, 409, 412} for trace in commit_traces + ), commit_traces + case.s3_proxy.clear_faults() + case.dynamodb_proxy.clear_faults() + health = check_dataset_in_fresh_process(case.uri, case.storage_options) + assert_history_conforms(health, foreign_writer_history_contract(operation)) + assert result_a.status in {"success", "error"}, result_a + assert result_b.status in {"success", "error"}, result_b + assert "success" in {result_a.status, result_b.status} + assert_successful_result_is_visible(result_a, health, operation, "a") + assert_successful_result_is_visible(result_b, health, operation, "b") + assert_results_match_visible_intents( + (result_a, result_b), health, operation + ) + finally: + stop_operation(writer_a) + stop_operation(writer_b) + persist_case_traces( + case, + tmp_path, + "foreign-writer", + metadata={ + "writer_a": asdict(result_a) if result_a is not None else None, + "writer_b": asdict(result_b) if result_b is not None else None, + }, + ) + + +@pytest.mark.integration +@pytest.mark.recurring +@pytest.mark.parametrize("backend", BACKENDS) +@pytest.mark.parametrize("operation", OPERATIONS) +def test_fault_sweep_accepts_only_complete_states( + conformance_bucket: str, + conformance_dynamodb_table: str, + tmp_path: Path, + backend: CommitBackend, + operation: CommitOperation, +) -> None: + with make_case( + backend=backend, + operation=operation, + bucket=conformance_bucket, + dynamodb_table=conformance_dynamodb_table, + ) as baseline: + baseline_result = execute_operation( + baseline.uri, baseline.storage_options, operation + ) + assert baseline_result.status == "success", baseline_result + mutation_counts = { + "s3": sum(trace.operation != "" for trace in _mutations(baseline.s3_proxy)), + "dynamodb": sum( + trace.operation != "" for trace in _mutations(baseline.dynamodb_proxy) + ), + } + + for service, count in mutation_counts.items(): + for occurrence in range(1, count + 1): + for phase in ("before", "after"): + with make_case( + backend=backend, + operation=operation, + bucket=conformance_bucket, + dynamodb_table=conformance_dynamodb_table, + ) as case: + proxy = case.s3_proxy if service == "s3" else case.dynamodb_proxy + plan = FaultPlan( + operation="mutation", + occurrence=occurrence, + phase=phase, + response_losses=100 if phase == "before" else 1, + verification_response_losses=100, + ) + proxy.arm(plan) + scenario = f"sweep-{service}-{occurrence}-{phase}" + result: OperationResult | None = None + try: + result = execute_operation( + case.uri, case.storage_options, operation + ) + assert any(trace.injected for trace in proxy.traces()), ( + service, + occurrence, + phase, + proxy.traces(), + ) + case.s3_proxy.clear_faults() + case.dynamodb_proxy.clear_faults() + health = check_dataset_in_fresh_process( + case.uri, case.storage_options + ) + assert result.status in {"success", "error"}, result + assert_history_conforms( + health, + operation_history_contract( + operation, + latest=( + "after" if result.status == "success" else "either" + ), + ), + ) + assert_successful_result_is_visible(result, health, operation) + finally: + persist_case_traces( + case, + tmp_path, + scenario, + metadata={ + "operation_result": ( + asdict(result) if result is not None else None + ), + }, + ) + + +def _mutations(proxy: CommitFaultProxy) -> list[RequestTrace]: + return [ + trace + for trace in proxy.traces() + if trace.operation + in { + "s3_delete", + "s3_copy", + "s3_put", + "s3_post", + "s3_manifest_create", + "ddb_commit_create", + "ddb_commit_finalize", + "ddb_put", + "ddb_delete", + } + ] + + +@pytest.mark.integration +@pytest.mark.recurring +@pytest.mark.parametrize("backend", BACKENDS) +@pytest.mark.parametrize("operation", OPERATIONS) +@pytest.mark.parametrize("durable_commit", [False, True]) +def test_unknown_outcome_recovery_applies_logical_operation_once( + conformance_bucket: str, + conformance_dynamodb_table: str, + tmp_path: Path, + backend: CommitBackend, + operation: CommitOperation, + durable_commit: bool, +) -> None: + with make_case( + backend=backend, + operation=operation, + bucket=conformance_bucket, + dynamodb_table=conformance_dynamodb_table, + ) as case: + plan = FaultPlan( + operation=commit_request_operation(backend), + occurrence=commit_request_occurrence(operation), + phase="after" if durable_commit else "before", + response_losses=1 if durable_commit else 100, + verification_response_losses=100 if durable_commit else 0, + ) + case.commit_proxy.arm(plan) + result: OperationResult | None = None + try: + result = execute_operation(case.uri, case.storage_options, operation) + case.s3_proxy.clear_faults() + case.dynamodb_proxy.clear_faults() + recovery, health = recover_unknown_operation( + case.uri, case.storage_options, operation + ) + assert recovery == ("already_applied" if durable_commit else "retried") + assert_history_conforms( + health, + operation_history_contract( + operation, + latest="after", + attempts=2 if recovery == "retried" else 1, + ), + ) + assert result.status == "error", result + finally: + persist_case_traces( + case, + tmp_path, + f"recover-{'after' if durable_commit else 'before'}", + metadata={ + "operation_result": asdict(result) if result is not None else None, + }, + ) + + +@pytest.mark.integration +@pytest.mark.recurring +@pytest.mark.parametrize("backend", BACKENDS) +@pytest.mark.parametrize("operation", OPERATIONS) +def test_gc_after_unknown_outcome_preserves_committed_artifacts( + conformance_bucket: str, + conformance_dynamodb_table: str, + tmp_path: Path, + backend: CommitBackend, + operation: CommitOperation, +) -> None: + with make_case( + backend=backend, + operation=operation, + bucket=conformance_bucket, + dynamodb_table=conformance_dynamodb_table, + ) as case: + plan = FaultPlan( + operation=commit_request_operation(backend), + occurrence=commit_request_occurrence(operation), + verification_response_losses=100, + ) + case.commit_proxy.arm(plan) + result: OperationResult | None = None + try: + result = execute_operation(case.uri, case.storage_options, operation) + case.s3_proxy.clear_faults() + case.dynamodb_proxy.clear_faults() + cleanup = execute_operation(case.uri, case.storage_options, "cleanup") + assert cleanup.status == "success", cleanup + health = check_dataset_in_fresh_process(case.uri, case.storage_options) + assert_history_conforms( + health, operation_history_contract(operation, latest="after") + ) + assert result.status == "error", result + finally: + persist_case_traces( + case, + tmp_path, + "gc-after-unknown", + metadata={ + "operation_result": asdict(result) if result is not None else None, + }, + ) + + +@pytest.mark.integration +@pytest.mark.recurring +@pytest.mark.parametrize("backend", BACKENDS) +@pytest.mark.parametrize("operation", OPERATIONS) +def test_process_exit_after_durable_commit_preserves_state( + conformance_bucket: str, + conformance_dynamodb_table: str, + tmp_path: Path, + backend: CommitBackend, + operation: CommitOperation, +) -> None: + with make_case( + backend=backend, + operation=operation, + bucket=conformance_bucket, + dynamodb_table=conformance_dynamodb_table, + ) as case: + gate = tmp_path / f"kill-start-{backend}-{operation}" + process = start_operation( + case.uri, + case.storage_options, + operation, + start_gate=gate, + ) + plan = FaultPlan( + operation=commit_request_operation(backend), + occurrence=commit_request_occurrence(operation), + kill_pid=process.pid, + ) + case.commit_proxy.arm(plan) + result: OperationResult | None = None + try: + gate.touch() + result = finish_operation(process) + case.s3_proxy.clear_faults() + case.dynamodb_proxy.clear_faults() + health = check_dataset_in_fresh_process(case.uri, case.storage_options) + assert_history_conforms( + health, operation_history_contract(operation, latest="after") + ) + assert result.status == "killed", result + finally: + stop_operation(process) + persist_case_traces( + case, + tmp_path, + "process-exit", + metadata={ + "operation_result": asdict(result) if result is not None else None, + }, + ) + + +@pytest.mark.integration +@pytest.mark.recurring +def test_fault_plan_trace_is_replayable( + conformance_bucket: str, + conformance_dynamodb_table: str, + tmp_path: Path, +) -> None: + original = FaultPlan(operation="s3_manifest_create", occurrence=1) + plans = [original] + normalized_traces: list[list[tuple[object, ...]]] = [] + for run in (1, 2): + plan = plans[-1] + with make_case( + backend="conditional", + operation="append", + bucket=conformance_bucket, + dynamodb_table=conformance_dynamodb_table, + ) as case: + case.s3_proxy.arm(plan) + try: + result = execute_operation(case.uri, case.storage_options, "append") + assert result.status == "success", result + operation_traces = case.s3_proxy.traces() + case.s3_proxy.clear_faults() + health = check_dataset_in_fresh_process(case.uri, case.storage_options) + assert_history_conforms( + health, operation_history_contract("append", latest="after") + ) + assert_successful_result_is_visible(result, health, "append") + normalized_traces.append( + [ + ( + trace.operation, + trace.forwarded, + trace.upstream_status, + trace.client_status, + trace.injected, + trace.fault_phase, + ) + for trace in operation_traces + ] + ) + finally: + persist_case_traces(case, tmp_path, f"replay-{run}") + if run == 1: + artifact = trace_directory(tmp_path) / "conditional-append-replay-1-s3.json" + replayed = read_fault_plans(artifact) + assert replayed == [original] + plans.append(replayed[0]) + assert normalized_traces[0] == normalized_traces[1] + + +@pytest.mark.integration +@pytest.mark.recurring +def test_fresh_process_oracle_detects_missing_index_artifact( + conformance_bucket: str, + conformance_dynamodb_table: str, + tmp_path: Path, +) -> None: + with make_case( + backend="conditional", + operation="create_index", + bucket=conformance_bucket, + dynamodb_table=conformance_dynamodb_table, + ) as case: + result = execute_operation(case.uri, case.storage_options, "create_index") + assert result.status == "success", result + check_dataset_in_fresh_process(case.uri, case.storage_options) + + parsed = urlsplit(case.uri) + prefix = parsed.path.lstrip("/") + s3 = boto3_client("s3", endpoint_url=S3_ENDPOINT) + objects = s3.list_objects_v2(Bucket=parsed.netloc, Prefix=prefix).get( + "Contents", [] + ) + index_artifacts = [ + item["Key"] for item in objects if "/_indices/" in item["Key"] + ] + assert index_artifacts, objects + s3.delete_object(Bucket=parsed.netloc, Key=index_artifacts[0]) + + with pytest.raises(AssertionError, match="fresh-process dataset validation"): + check_dataset_in_fresh_process(case.uri, case.storage_options) + persist_case_traces(case, tmp_path, "negative-missing-index-artifact") + + +@pytest.mark.integration +@pytest.mark.recurring +@pytest.mark.real_s3 +@pytest.mark.parametrize("operation", OPERATIONS) +@pytest.mark.parametrize("lose_verification_reads", [False, True]) +def test_real_s3_commit_response_loss( + tmp_path: Path, + operation: CommitOperation, + lose_verification_reads: bool, +) -> None: + bucket = os.environ.get("LANCE_CONFORMANCE_REAL_S3_BUCKET") + if bucket is None: + pytest.skip("LANCE_CONFORMANCE_REAL_S3_BUCKET is not configured") + if os.environ.get("LANCE_CONFORMANCE_REAL_S3_ISOLATED") != "1": + pytest.skip("real-S3 faults must be run through the fail-closed CI runner") + + region = os.environ.get("LANCE_CONFORMANCE_REAL_S3_REGION", "us-east-2") + listen_port = int(os.environ.get("LANCE_CONFORMANCE_PROXY_PORT", "18080")) + upstream_connect_host = os.environ["LANCE_CONFORMANCE_REAL_S3_UPSTREAM_IP"] + upstream = f"https://s3.{region}.amazonaws.com" + client_endpoint = f"http://s3.{region}.amazonaws.com" + uri = f"s3://{bucket}/commit-conformance/{uuid.uuid4().hex}.lance" + options = { + "allow_http": "true", + "aws_region": region, + "aws_endpoint": client_endpoint, + "aws_virtual_hosted_style_request": "false", + "client_max_retries": "2", + "client_retry_timeout": "5", + } + + parsed = urlsplit(uri) + s3 = boto3.client( + "s3", + region_name=region, + endpoint_url=client_endpoint, + config=Config(s3={"addressing_style": "path"}), + ) + with CommitFaultProxy( + upstream, + listen_port=listen_port, + upstream_connect_host=upstream_connect_host, + ) as proxy: + plan: FaultPlan | None = None + result: OperationResult | None = None + try: + setup = execute_operation(uri, options, "setup") + assert setup.status == "success", ( + "real-S3 transparent proxy is not reachable; configure the host's " + f"port-80 redirect to 127.0.0.1:{listen_port}", + setup, + ) + assert any( + trace.operation == "s3_manifest_create" for trace in proxy.traces() + ), ( + "real-S3 setup bypassed the transparent fault proxy; the test " + "cannot prove that response loss was injected", + proxy.traces(), + ) + proxy.clear_traces() + plan = FaultPlan( + operation="s3_manifest_create", + occurrence=commit_request_occurrence(operation), + verification_response_losses=100 if lose_verification_reads else 0, + ) + proxy.arm(plan) + result = execute_operation(uri, options, operation) + traces = proxy.traces() + assert any( + trace.operation == "s3_manifest_create" + and trace.injected + and trace.fault_phase == "after" + for trace in traces + ), ( + "the real-S3 commit response-loss fault was not exercised", + traces, + ) + assert any(trace.operation == "s3_manifest_read" for trace in traces), ( + "the conditional commit did not perform read-back after the " + "lost response", + traces, + ) + if lose_verification_reads: + assert any( + trace.operation == "s3_manifest_read" + and trace.injected + and trace.fault_phase == "verification" + for trace in traces + ), ( + "the real-S3 verification response-loss fault was not exercised", + traces, + ) + proxy.clear_faults() + health = check_dataset_in_fresh_process(uri, options) + assert_history_conforms( + health, operation_history_contract(operation, latest="after") + ) + expected_status = "error" if lose_verification_reads else "success" + assert result.status == expected_status, result + assert_successful_result_is_visible(result, health, operation) + finally: + proxy.clear_faults() + if plan is not None: + proxy.write_trace( + trace_directory(tmp_path) + / f"real-s3-{operation}-{lose_verification_reads}.json", + metadata={ + "backend": "real-s3", + "operation": operation, + "lose_verification_reads": lose_verification_reads, + "operation_result": ( + asdict(result) if result is not None else None + ), + }, + ) + objects = s3.list_objects_v2( + Bucket=bucket, Prefix=parsed.path.lstrip("/") + ).get("Contents", []) + if objects: + s3.delete_objects( + Bucket=bucket, + Delete={"Objects": [{"Key": item["Key"]} for item in objects]}, + ) + + +def test_final_state_oracle_rejects_partial_and_duplicate_states() -> None: + before = base_state() + after = state_after_operation("append") + + def health_for(states: list[DatasetState]) -> DatasetHealth: + versions = list(range(1, len(states) + 1)) + return { + "versions": versions, + "opened_version": versions[-1], + "states": {str(version): state for version, state in zip(versions, states)}, + "transactions": { + str(version): { + "uuid": f"transaction-{version}", + "operation": "TestOperation", + } + for version in versions + }, + } + + contract = operation_history_contract("append", latest="after") + assert_history_conforms(health_for([before, after]), contract) + + duplicate = json.loads(json.dumps(after)) + duplicate["rows"].append({"id": 200}) + with pytest.raises(AssertionError): + assert_history_conforms(health_for([before, duplicate]), contract) + + partial = json.loads(json.dumps(after)) + partial["rows"] = partial["rows"][:-1] + with pytest.raises(AssertionError): + assert_history_conforms(health_for([before, partial]), contract) + + with pytest.raises(AssertionError, match="too many versions"): + assert_history_conforms(health_for([before, after, after]), contract) + + with pytest.raises(AssertionError): + assert_history_conforms(health_for([before, after, before]), contract) + + stale_open = health_for([before, after]) + stale_open["opened_version"] = 1 + with pytest.raises(AssertionError, match="latest visible version"): + assert_history_conforms(stale_open, contract) + + duplicate_transaction = health_for([before, after]) + duplicate_transaction["transactions"]["2"]["uuid"] = "transaction-1" + with pytest.raises(AssertionError, match="committed more than once"): + assert_history_conforms(duplicate_transaction, contract) + + compaction_retry = health_for( + [before, before, before, state_after_operation("compaction")] + ) + assert_history_conforms( + compaction_retry, + operation_history_contract("compaction", latest="after", attempts=2), + ) + with pytest.raises(AssertionError, match="too many versions"): + assert_history_conforms( + compaction_retry, + operation_history_contract("compaction", latest="after"), + ) + + compacted = state_after_operation("compaction") + concurrent_compaction = health_for([before, before, compacted, compacted]) + concurrent_compaction["transactions"]["1"]["operation"] = "Overwrite" + concurrent_compaction["transactions"]["2"]["operation"] = "BaseOperation" + concurrent_compaction["transactions"]["3"]["operation"] = "Rewrite" + concurrent_compaction["transactions"]["4"]["operation"] = "BaseOperation" + assert_history_conforms( + concurrent_compaction, foreign_writer_history_contract("compaction") + ) + compaction_result = OperationResult( + status="success", + returncode=0, + error_type=None, + error_message=None, + committed_version=3, + transaction_uuid="transaction-3", + stdout="", + stderr="", + ) + assert_results_match_visible_intents( + (compaction_result,), concurrent_compaction, "compaction" + ) + + duplicate_rewrite = json.loads(json.dumps(concurrent_compaction)) + duplicate_rewrite["transactions"]["4"]["operation"] = "Rewrite" + with pytest.raises(AssertionError, match="visible commit intents"): + assert_results_match_visible_intents( + (compaction_result,), duplicate_rewrite, "compaction" + ) + + +def test_local_oracle_correlates_success_with_visible_transaction( + tmp_path: Path, +) -> None: + uri = str(tmp_path / "local-oracle.lance") + setup = execute_operation(uri, {}, "setup") + assert setup.status == "success", setup + result = execute_operation(uri, {}, "append") + assert result.status == "success", result + + health = check_dataset_in_fresh_process(uri, {}) + assert_history_conforms( + health, operation_history_contract("append", latest="after") + ) + assert_successful_result_is_visible(result, health, "append") + + +def test_operation_timeout_reaps_child() -> None: + process = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(60)"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + with pytest.raises(AssertionError, match="timed out"): + finish_operation(process, timeout=0.01) + assert process.poll() is not None + + +def test_trace_rebinds_process_kill_fault(tmp_path: Path) -> None: + trace = tmp_path / "kill-fault.json" + with CommitFaultProxy("http://127.0.0.1:9") as proxy: + proxy.arm(FaultPlan(operation="mutation", kill_pid=12345)) + proxy.write_trace(trace) + + assert '"kill_pid"' not in trace.read_text() + with pytest.raises(ValueError, match="requires a new kill_pid"): + read_fault_plans(trace) + replayed = read_fault_plans(trace, kill_pid=54321) + assert replayed[0].kill_pid == 54321 diff --git a/rust/lance/src/io/commit.rs b/rust/lance/src/io/commit.rs index f809a23a9dd..88a865b85dc 100644 --- a/rust/lance/src/io/commit.rs +++ b/rust/lance/src/io/commit.rs @@ -159,9 +159,36 @@ enum CommitOutcome { /// Maximum verification read attempts in [`verify_commit_outcome`]. const COMMIT_VERIFICATION_ATTEMPTS: u32 = 3; +/// Validate the identity used to prove ownership after an ambiguous commit. +/// +/// Transaction UUIDs are normally generated by +/// [`crate::dataset::transaction::TransactionBuilder`]. Public bindings can +/// also construct transactions directly, so enforce the invariant at the +/// common commit boundary before writing any artifacts. +fn validate_transaction_identity(transaction: &Transaction) -> Result<()> { + if transaction.uuid.is_empty() { + return Err(Error::invalid_input( + "Transaction UUID must not be empty because it identifies the commit attempt", + )); + } + Ok(()) +} + +/// Compare both the stable attempt identity and its canonical persisted intent. +/// +/// The protobuf representation is the transaction's persistence contract. It +/// normalizes fields such as index timestamps, so comparing it avoids false +/// negatives after a round trip while preventing a reused UUID from claiming a +/// different transaction. +fn is_same_commit_attempt(committed: &Transaction, attempted: &Transaction) -> bool { + !attempted.uuid.is_empty() + && committed.uuid == attempted.uuid + && pb::Transaction::from(committed) == pb::Transaction::from(attempted) +} + /// Determine whether a failed commit attempt actually landed, by comparing -/// the complete transaction recorded in the manifest at `version` with this -/// attempt's transaction. +/// the stable identity and canonical persisted intent recorded in the manifest +/// at `version` with this attempt's transaction. /// /// Never returns an error. Read failures and non-definitive not-found results /// are retried briefly, then collapse to [`CommitOutcome::Unknown`]. @@ -186,7 +213,7 @@ async fn verify_commit_outcome( match read_manifest_transaction(object_store, base_path, &manifest, &location).await { Ok(Some(committed_transaction)) => { - return if committed_transaction == *transaction { + return if is_same_commit_attempt(&committed_transaction, transaction) { CommitOutcome::Ours { manifest: Box::new(manifest), location, @@ -316,6 +343,8 @@ async fn do_commit_new_dataset( metadata_cache: &DSMetadataCache, store_registry: Arc, ) -> Result<(Manifest, ManifestLocation)> { + validate_transaction_identity(transaction)?; + let transaction_file = if !write_config.disable_transaction_file() { write_transaction_file(object_store, base_path, transaction).await? } else { @@ -1040,6 +1069,8 @@ pub(crate) async fn do_commit_detached_transaction( write_config: &ManifestWriteConfig, commit_config: &CommitConfig, ) -> Result<(Manifest, ManifestLocation)> { + validate_transaction_identity(transaction)?; + // We don't strictly need a transaction file but we go ahead and create one for // record-keeping if nothing else. let transaction_file = if !write_config.disable_transaction_file() { @@ -1280,6 +1311,8 @@ pub(crate) async fn commit_transaction( manifest_naming_scheme: ManifestNamingScheme, affected_rows: Option<&RowAddrTreeMap>, ) -> Result<(Manifest, ManifestLocation)> { + validate_transaction_identity(transaction)?; + // Note: object_store has been configured with WriteParams, but dataset.object_store.as_ref() // has not necessarily. So for anything involving writing, use `object_store`. let read_version = transaction.read_version; @@ -1738,6 +1771,43 @@ mod tests { assert_eq!(transaction.tag, read_transaction.tag); } + #[test] + fn test_commit_attempt_match_requires_identity_and_intent() { + use crate::dataset::transaction::TransactionBuilder; + + let attempted = TransactionBuilder::new(7, Operation::Append { fragments: vec![] }) + .uuid("shared-attempt-id".to_string()) + .tag(Some("attempted".to_string())) + .build(); + let persisted = pb::Transaction::from(&attempted).try_into().unwrap(); + assert!(is_same_commit_attempt(&persisted, &attempted)); + + let different_intent = TransactionBuilder::new( + 7, + Operation::Delete { + updated_fragments: vec![], + deleted_fragment_ids: vec![], + predicate: "false".to_string(), + }, + ) + .uuid(attempted.uuid.clone()) + .tag(attempted.tag.clone()) + .build(); + assert!( + !is_same_commit_attempt(&different_intent, &attempted), + "a reused UUID must not claim a different transaction" + ); + + let missing_identity = TransactionBuilder::new(7, Operation::Append { fragments: vec![] }) + .uuid(String::new()) + .build(); + assert!(!is_same_commit_attempt( + &missing_identity, + &missing_identity + )); + assert!(validate_transaction_identity(&missing_identity).is_err()); + } + #[tokio::test] async fn test_concurrent_create_index() { // Create a table with two vector columns @@ -2377,6 +2447,47 @@ mod tests { assert_eq!(count_txn_files(uri), txn_files_before + 1); } + /// CreateIndex metadata is normalized by its protobuf round trip, so + /// ownership must compare the attempt identity and canonical persisted + /// intent instead of the decoded domain structs. + #[tokio::test] + async fn test_create_index_succeeds_when_conflict_is_own_commit() { + use crate::utils::test::{AmbiguousCommitHandler, AmbiguousFailure}; + use lance_index::scalar::ScalarIndexParams; + + let tmp = TempStrDir::default(); + let uri = tmp.as_str(); + let schema = simple_schema(); + let handler = Arc::new(AmbiguousCommitHandler::default()); + let params = WriteParams { + commit_handler: Some(handler.clone()), + ..Default::default() + }; + let reader = RecordBatchIterator::new( + vec![Ok(simple_batch(&schema, vec![1, 2, 3]))], + schema.clone(), + ); + let mut dataset = Dataset::write(reader, uri, Some(params)).await.unwrap(); + + handler.fail_next(AmbiguousFailure::LandAndConflict); + dataset + .create_index( + &["x"], + IndexType::BTree, + Some("x_idx".to_string()), + &ScalarIndexParams::default(), + true, + ) + .await + .expect("a landed CreateIndex commit must be recognized as our attempt"); + + let fresh = Dataset::open(uri).await.unwrap(); + assert_eq!(fresh.version().version, 2); + let indices = fresh.load_indices().await.unwrap(); + assert_eq!(indices.len(), 1); + assert_eq!(indices[0].name, "x_idx"); + } + #[tokio::test] async fn test_commit_retries_temporarily_invisible_manifest() { use crate::utils::test::{AmbiguousCommitHandler, AmbiguousFailure};