diff --git a/.github/workflows/rustfs-s3-compat-test.yml b/.github/workflows/rustfs-s3-compat-test.yml index aa5a6bc..a9bf1b4 100644 --- a/.github/workflows/rustfs-s3-compat-test.yml +++ b/.github/workflows/rustfs-s3-compat-test.yml @@ -77,6 +77,14 @@ jobs: openssl version df -h /data | tail -1 + - name: Install bulk-delete regression client + run: | + set -euo pipefail + BULK_CLIENT="${FUNCTIONAL_ARTIFACTS_DIR}-scratch/bulk-client" + python3 -m venv "${BULK_CLIENT}" + "${BULK_CLIENT}/bin/python" -m pip install 'boto3>=1.35,<2' + printf 'RUSTFS_S3_BULK_PYTHON=%s/bin/python\n' "${BULK_CLIENT}" >> "${GITHUB_ENV}" + - name: Admit clean shared environment id: environment timeout-minutes: 5 @@ -113,7 +121,7 @@ jobs: chmod +x rustfs-s3-compat-test.sh PACKAGE_URL='${{ inputs.package_url }}' RUSTFS_VERSION='${{ inputs.rustfs_version }}' - ARGS=(--all-topologies -y --log-file "${LOG_FILE}") + ARGS=(--all-topologies -y --log-file "${LOG_FILE}" --bulk-delete-results-dir "${FUNCTIONAL_ARTIFACTS_DIR}/bulk-delete") if [ -n "${PACKAGE_URL}" ]; then ARGS+=(--package-url "${PACKAGE_URL}") elif [ -n "${RUSTFS_VERSION}" ]; then @@ -292,6 +300,7 @@ jobs: ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/report.md ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/suite.log ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/cases.md + ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/bulk-delete/ if-no-files-found: error - name: Cleanup environment (after) diff --git a/.github/workflows/s3-script-tests.yml b/.github/workflows/s3-script-tests.yml index a51b9d8..4b33006 100644 --- a/.github/workflows/s3-script-tests.yml +++ b/.github/workflows/s3-script-tests.yml @@ -8,6 +8,9 @@ on: - test_rustfs_s3_usage.py - test_rustfs_s3_signing.py - test_rustfs_s3_regressions.py + - scripts/s3_bulk_delete_regressions.py + - scripts/performance_s3.py + - test_s3_bulk_delete_regressions.py - test_rustfs_worm_object_lock.py - test_validation_s3_basic.sh - .github/workflows/s3-script-tests.yml @@ -19,6 +22,9 @@ on: - test_rustfs_s3_usage.py - test_rustfs_s3_signing.py - test_rustfs_s3_regressions.py + - scripts/s3_bulk_delete_regressions.py + - scripts/performance_s3.py + - test_s3_bulk_delete_regressions.py - test_rustfs_worm_object_lock.py - test_validation_s3_basic.sh - .github/workflows/s3-script-tests.yml @@ -37,8 +43,8 @@ jobs: - name: Validate shell syntax run: bash -n rustfs-s3-compat-test.sh scripts/campaign/run-worm-object-lock.sh - name: Install independent SigV4 verifier - run: python3 -m pip install botocore==1.43.19 + run: python3 -m pip install botocore==1.43.19 'boto3>=1.43.19,<1.43.20' - name: Test S3 cases and signatures without a cluster - run: python3 -m unittest test_rustfs_s3_usage.py test_rustfs_s3_signing.py test_rustfs_s3_regressions.py test_rustfs_worm_object_lock.py + run: python3 -B -m unittest test_rustfs_s3_usage.py test_rustfs_s3_signing.py test_rustfs_s3_regressions.py test_rustfs_worm_object_lock.py test_s3_bulk_delete_regressions.py - name: Test validation S3 round trip without a cluster run: bash test_validation_s3_basic.sh diff --git a/README.md b/README.md index 15e783f..8871dbc 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,10 @@ per-key failures, request ids, GET results and residue before/after cleanup. Cleanup runs in `finally`; an incomplete cleanup aborts the remaining cases. The caller must hold the environment's test lock throughout this probe. +## Bulk-delete regressions + +The complete S3 topology suite includes bulk-delete regressions `S3C-213`, `S3C-214`, `S3C-1306` and `S3C-1307`. They cover batch boundaries, duplicate keys, immutable GET load and finite same-key PUT contention in off/on/suspended versioning modes. Python boto3 is required. Each subcase retains JSON, phase/request timing, HTTP 503 and per-object error counts, version/marker cleanup receipts and a Chinese Markdown summary under `--bulk-delete-results-dir`. The standalone runner accepts existing endpoints and does not deploy nodes; its caller must hold the environment lock. See [the case contracts and usage](docs/s3-bulk-delete-regressions.zh-CN.md). + ## Release validation migration `validation-cases.json` snapshots all 158 canonical cases from diff --git a/docs/s3-bulk-delete-regressions.zh-CN.md b/docs/s3-bulk-delete-regressions.zh-CN.md new file mode 100644 index 0000000..7979f10 --- /dev/null +++ b/docs/s3-bulk-delete-regressions.zh-CN.md @@ -0,0 +1,47 @@ +# 批量删除回归用例 + +这些用例来自 main `0f9258594` 全量复测和后续批量删除诊断中发现的逐对象 `ServiceUnavailable`、准备/清理超时及热点并发问题。用例验证 S3 可观察行为；RPC 内部的迟到响应、exact-ID 清理和按键重试仍由 RustFS 锁库的回归测试验证，不能仅凭外部请求成功声称内部缺陷已修复。 + +## 用例与判定 + +| 用例 | 子场景 | 通过条件 | +|---|---|---| +| S3C-213 BulkDeleteBatchMatrix | 版本关闭/开启/暂停 × 1、32、128、500、1000 键，共 15 项 | 一次请求中每条输入均有结果、无逐对象错误；各端点均看不到已删当前对象；未参与删除的对象字节不变；开启和暂停模式下保留抽样历史版本的原始字节。 | +| S3C-214 BulkDeleteDuplicateKeys | 三种版本状态，共 3 项；包含同一键重复 16 次和两个重复的不存在键 | 按输入条目数核对响应，不把重复条目当作独立对象；当前键删除且控制对象、历史版本保持正确。 | +| S3C-1306 BulkDeleteUnderGetLoad | 三种版本状态，共 3 项；默认 32 个读取工作线程和 1000 键删除 | 所有线程先成功读取控制对象；并发负载期间读取字节保持一致；一次批删无错误、无当前对象残留；所有传入端点均验证结果。 | +| S3C-1307 BulkDeleteWriteRecovery | 三种版本状态，共 3 项；默认 32 个有限写入线程、一个热点键和 127 个其他键，热点键在批删输入中重复 16 次 | 原始请求及负载无错误、其他键删除；竞争结束时热点值不存在或为完整已写入值，各端点视图一致；随后同键 PUT、各端点 GET、DELETE 及其他键重试成功，最终无当前对象残留。 | + +默认每个拓扑执行 24 个子场景，全量 S3 套件的三个拓扑共执行 72 个子场景。对象体为 128 KiB，最大单批种子数据约 125 MiB，不复用压测中的历史大桶。每个子场景使用新建的独立桶，清理显式移除对象版本、删除标记及未完成分片上传，再检查没有残留并删除桶。不会删除其他桶、修改节点配置或注入环境故障。 + +原始批删只发起一次请求，SDK 自动重试关闭。HTTP 200 的 `DeleteResult/Errors` 是失败；顶层 HTTP 503 响应数和逐对象 `ServiceUnavailable` 条目数分别记录，不能互相换算。恢复或清理重试成功不会覆盖原始失败，清理曾出现错误的子场景也保持 FAIL。一次调用中清理最终失败或中断时，该调用的后续子场景记录 NOT_RUN，保留责任与证据；全量拓扑套件其他独立用例沿用原有调度。 + +热点用例在记录原始请求失败后继续验证无负载恢复，分别保留原始失败和恢复结果。故意制造高负载下的合法仲裁超时仍会使严格回归判定失败；失败结果是需要调查的现象，不能自动认定每个 503 都是同一种产品缺陷。 + +两个负载场景均在所有工作线程完成预热后释放后续请求，并等待至少一个实际 SDK 调用进入，再发起原始批删。只有预热之后成功的 GET/PUT 调用与批删调用在客户端的单调时钟区间存在正交集，负载条件才通过；仅预热成功、所有请求提前完成或重叠请求全部失败都不能通过。GET 的区间包含响应体读取，客户端创建、PUT 响应后的日志写入和批删响应完整性核对不计入请求区间。结果记录重叠成功次数、工作线程数和首个交集，事件文件保留每次交集；这些是客户端并发证据，不是服务端锁重叠的证明。 + +HTTP 200 响应即使漏掉部分输入，也会先保存已解析的删除确认和逐对象错误，再以 `IncompleteDeleteResponse` 判失败。恢复阶段自身失败时保留原始阶段的独立结果；原始验证中的流式读取超时或短读也会记录为失败，然后继续热点键的无负载恢复检查。 + +## 执行与证据 + +全量入口为 `rustfs-s3-compat-test.sh --all-topologies`，也可以使用 `--cases S3C-213,S3C-214,S3C-1306,S3C-1307` 选择这些用例。`--bulk-delete-results-dir DIR` 指定持久证据目录；未指定时创建独立目录，有日志文件则放在日志旁边。多节点拓扑使用既有入口和配置的其他节点地址直接验证。S3 workflow 在本次任务的独立 venv 中安装 boto3，避免修改宿主 Python 包，并自动上传结果目录；手工执行时可通过 `RUSTFS_S3_BULK_PYTHON` 指定已安装 boto3 的 Python 解释器。 + +已经部署的集群可以直接运行以下命令。凭据使用 `RUSTFS_ACCESS_KEY`/`RUSTFS_SECRET_KEY` 或 `AWS_ACCESS_KEY_ID`/`AWS_SECRET_ACCESS_KEY` 环境变量；令牌对应 `RUSTFS_SESSION_TOKEN` 或 `AWS_SESSION_TOKEN`。调用方必须在整个测试和清理期间持有环境共享锁。 + +```bash +python3 scripts/s3_bulk_delete_regressions.py \ + --endpoint http://node1:9000 \ + --endpoint http://node2:9000 \ + --endpoint http://node3:9000 \ + --endpoint http://node4:9000 \ + --result-dir /absolute/path/to/new-results +``` + +默认准备预算 300 秒，请求/验证/恢复各阶段预算 60 秒，负载预算 30 秒，清理预算 180 秒；通过 `--prepare-budget`、`--request-budget`、`--load-budget`、`--cleanup-budget` 调整。每阶段各端点共用一个截止时间，不允许后续请求重新获得完整预算。客户端连接和读取设置有界超时，负载线程执行有限轮次；网络读取超时是 socket 读取超时，不应解读为逐字节流的硬墙钟限制。 + +每个负载线程复用自己的客户端与连接池；剩余预算不足其 10 秒读取超时后才重建并收紧超时。共享性能工具仍保持原有 60 秒读取超时。 + +每个子场景保留 `result.json`、`s3-events.jsonl` 和 `bucket.s3.json`。结果包含第一失败阶段、后续失败列表、各阶段耗时、请求 ID、SDK 重试次数、各阶段 HTTP 503 响应数、逐对象错误条目、内容校验和清理结果。`summary.json` 和中文 `summary.md` 汇总全部计划子场景及未执行项。结果目录必须全新，禁止用后续运行覆盖原始证据。 + +## 本地验证范围 + +`python3 -B -m unittest test_s3_bulk_delete_regressions.py` 使用响应变异和版本模型验证通过/失败判定、重复输入计数、历史版本保留、泄漏检测、失败后恢复、阶段预算、清理失败历史、中断证据和全量入口选择。这些测试验证测试工具本身，不等同于 RustFS 测试机复测通过。性能准备/清理中断和报告前后超时的既有回归继续位于 `test_rustfs_performance_testing.py` 与 `test_performance_s3.py`，避免重复编写同一套模拟。 diff --git a/rustfs-s3-compat-test.sh b/rustfs-s3-compat-test.sh index 701a875..d54649f 100755 --- a/rustfs-s3-compat-test.sh +++ b/rustfs-s3-compat-test.sh @@ -94,6 +94,8 @@ VALIDATION_OBJECT_CLEANUP_STATUS=PASS VALIDATION_BINARY_SHA256="" ENDPOINT="" VOLUMES="" +S3_BULK_RESULTS_DIR="${RUSTFS_S3_BULK_RESULTS_DIR:-}" +S3_BULK_PYTHON="${RUSTFS_S3_BULK_PYTHON:-python3}" # ==================== Helpers ==================== @@ -882,6 +884,27 @@ worm_object_lock_tests() { run_test S3C-605 "LegalHold" test_s3c605 } +bulk_delete_regression() { + local id="$1" index=1 node + local args=(--case "${id}" --endpoint "${ENDPOINT}") + if [ -z "${S3_BULK_RESULTS_DIR}" ]; then + S3_BULK_RESULTS_DIR="$(mktemp -d "${LOG_FILE:+$(dirname "${LOG_FILE}")/}rustfs-s3-bulk.XXXXXX")" + fi + mkdir -p "${S3_BULK_RESULTS_DIR}/${CURRENT_TOPOLOGY:-existing}" + args+=(--result-dir "${S3_BULK_RESULTS_DIR}/${CURRENT_TOPOLOGY:-existing}/${id}") + for node in $(topology_nodes "${CURRENT_TOPOLOGY}"); do + if [ "${index}" -gt 1 ]; then args+=(--endpoint "http://${node}:9000"); fi + index=$((index+1)) + done + RUSTFS_ACCESS_KEY="${ACCESS_KEY}" RUSTFS_SECRET_KEY="${SECRET_KEY}" AWS_DEFAULT_REGION="${REGION}" \ + "${S3_BULK_PYTHON}" "${SCRIPT_DIR}/scripts/s3_bulk_delete_regressions.py" "${args[@]}" +} + +test_s3c213() { bulk_delete_regression S3C-213; } +test_s3c214() { bulk_delete_regression S3C-214; } +test_s3c1306() { bulk_delete_regression S3C-1306; } +test_s3c1307() { bulk_delete_regression S3C-1307; } + run_all_tests() { run_test S3C-101 "CreateBucket" test_s3c101 run_test S3C-102 "HeadBucket" test_s3c102 @@ -901,6 +924,8 @@ run_all_tests() { run_test S3C-210 "SpecialCharKey" test_s3c210 run_test S3C-211 "ServerSideMove" test_s3c211 run_test S3C-212 "ReservedStorageNames" test_s3c212 + run_test S3C-213 "BulkDeleteBatchMatrix" test_s3c213 + run_test S3C-214 "BulkDeleteDuplicateKeys" test_s3c214 run_test S3C-301 "MultipartFullFlow" test_s3c301 run_test S3C-302 "UploadPartCopy" test_s3c302 run_test S3C-303 "ListPartsAndUploads" test_s3c303 @@ -963,6 +988,8 @@ run_all_tests() { run_test S3C-1303 "HugeKeyMetadata" test_s3c1303 run_test S3C-1304 "MalformedXmlParams" test_s3c1304 run_test S3C-1305 "ConcurrentOverwrite" test_s3c1305 + run_test S3C-1306 "BulkDeleteUnderGetLoad" test_s3c1306 + run_test S3C-1307 "BulkDeleteWriteRecovery" test_s3c1307 run_test S3C-1401 "CopyMetadataOnCopy" test_s3c1401 run_test S3C-1402 "AwsChunkedNoDecodedLen" test_s3c1402 run_test S3C-1403 "ConcurrentUploadPart" test_s3c1403 @@ -3141,6 +3168,7 @@ Options: --ssh-user USER SSH user for the nodes (default azureuser) --ssh-port PORT SSH port for the nodes (default 22) --log-file FILE Append all output to FILE + --bulk-delete-results-dir DIR Retain bulk-delete JSON, request traces and Markdown --validation-case ID Run only s3.basic-object-round-trip (single-single) --validation-artifacts-dir DIR Write canonical validation result and evidence --candidate-manifest FILE Immutable candidate manifest for validation results @@ -3188,6 +3216,7 @@ main() { --ssh-user) SSH_USER="$1"; shift ;; --ssh-port) SSH_PORT="$1"; shift ;; --log-file) LOG_FILE="$1"; shift ;; + --bulk-delete-results-dir) S3_BULK_RESULTS_DIR="$1"; shift ;; --validation-case) VALIDATION_CASE="$1"; shift ;; --validation-artifacts-dir) VALIDATION_ARTIFACTS_DIR="$1"; shift ;; --candidate-manifest) VALIDATION_CANDIDATE_MANIFEST="$1"; shift ;; @@ -3247,6 +3276,13 @@ main() { need_cmd openssl "OpenSSL (SigV4)" need_cmd python3 "Python 3 (archive fixtures)" need_cmd tar "tar archive tool" + if [ "${DRY_RUN}" -eq 0 ] && [ -z "${VALIDATION_CASE}" ]; then + case ",${SELECTED_CASES}," in + ,,|*,S3C-213,*|*,S3C-214,*|*,S3C-1306,*|*,S3C-1307,*) + "${S3_BULK_PYTHON}" -c 'import boto3' >/dev/null 2>&1 || die "bulk-delete regressions require Python boto3" + ;; + esac + fi log "nodes: ${NODES[*]} package: $(resolve_package_url)" local overall_fail=0 topo for topo in "${SELECTED_TOPOLOGIES[@]}"; do diff --git a/scripts/performance_s3.py b/scripts/performance_s3.py index f063399..c1d9dc2 100644 --- a/scripts/performance_s3.py +++ b/scripts/performance_s3.py @@ -39,6 +39,8 @@ def new_bucket(prefix): class Session: + read_timeout = 60 + def __init__(self, endpoint, events, budget, factory=None): self.endpoint = endpoint self.events = events @@ -57,7 +59,7 @@ def client(self, remaining): aws_secret_access_key=os.environ["WARP_SECRET_KEY"], aws_session_token=os.environ.get("WARP_SESSION_TOKEN") or None, config=Config(s3={"addressing_style": "path"}, proxies={}, - connect_timeout=min(3, remaining), read_timeout=min(60, remaining), + connect_timeout=min(3, remaining), read_timeout=min(self.read_timeout, remaining), retries={"total_max_attempts": 1}), ) @@ -66,15 +68,20 @@ def emit(self, **record): with self.events.open("a") as stream: stream.write(json.dumps(record) + "\n") + def invoke(self, operation, **arguments): + return getattr(self.s3, operation)(**arguments) + def call(self, operation, **arguments): remaining = self.deadline - time.monotonic() if remaining <= 0: raise RequestFailure("CleanupDeadlineExceeded") try: # Rebuild near the deadline so a call cannot retain a longer old timeout. - if self.s3 is None or remaining < 60: + if self.s3 is None or remaining < self.read_timeout: self.s3 = self.factory(max(0.01, remaining)) - response = getattr(self.s3, operation)(**arguments) + response = self.invoke(operation, **arguments) + except RequestFailure: + raise except Exception as exc: response = getattr(exc, "response", {}) if not isinstance(response, dict): diff --git a/scripts/s3_bulk_delete_regressions.py b/scripts/s3_bulk_delete_regressions.py new file mode 100644 index 0000000..46f06f6 --- /dev/null +++ b/scripts/s3_bulk_delete_regressions.py @@ -0,0 +1,594 @@ +#!/usr/bin/env python3 +"""Run bounded bulk-delete regressions on an already deployed S3 cluster. + +The caller owns the environment lock. Only fresh, receipt-owned buckets are used. +""" +from __future__ import annotations + +import argparse +from collections import Counter +from concurrent.futures import ThreadPoolExecutor +import datetime +import hashlib +import json +import os +from pathlib import Path +import signal +import threading +import time +from urllib.parse import urlsplit + +try: + from .performance_s3 import Session, RequestFailure, RETRYABLE, abort_uploads, prepare, read_state, save +except ImportError: + from performance_s3 import Session, RequestFailure, RETRYABLE, abort_uploads, prepare, read_state, save + +GROUPS = {"S3C-213": "batch", "S3C-214": "duplicates", + "S3C-1306": "get-load", "S3C-1307": "put-delete"} +MODES = ("off", "on", "suspended") +PAYLOAD = (b"bulk-delete-regression\n" * 8192)[:131072] +CONTROL = "control/immutable" + + +def now(): + return datetime.datetime.now(datetime.timezone.utc).isoformat() + + +def require(ok, code): + if not ok: + raise RequestFailure(code) + + +def credentials(): + explicit = bool(os.environ.get("RUSTFS_ACCESS_KEY") or os.environ.get("RUSTFS_SECRET_KEY")) + access = os.environ.get("RUSTFS_ACCESS_KEY" if explicit else "AWS_ACCESS_KEY_ID") + secret = os.environ.get("RUSTFS_SECRET_KEY" if explicit else "AWS_SECRET_ACCESS_KEY") + require(access and secret, "MissingS3Credentials") + token = os.environ.get("RUSTFS_SESSION_TOKEN" if explicit else "AWS_SESSION_TOKEN") or None + return access, secret, token + + +class Requests(Session): + read_timeout = 10 + + def __init__(self, endpoint, events, budget, phase, factory=None): + super().__init__(endpoint, events, budget, factory) + self.phase = phase + self.window = None + self.worker = None + self.pending_read = None + + def client(self, remaining): + import boto3 + from botocore.config import Config + access, secret, token = credentials() + return boto3.client( + "s3", endpoint_url=self.endpoint, + region_name=os.environ.get("AWS_DEFAULT_REGION", "us-east-1"), + aws_access_key_id=access, aws_secret_access_key=secret, aws_session_token=token, + config=Config(s3={"addressing_style": "path"}, proxies={}, + connect_timeout=min(3, remaining), read_timeout=min(self.read_timeout, remaining), + retries={"total_max_attempts": 1}), + ) + + def emit(self, **record): + super().emit(phase=self.phase, **record) + + def invoke(self, operation, **arguments): + if self.window is None: + return super().invoke(operation, **arguments) + # Measure SDK calls after client construction and before response + # logging. A streamed GET stays active until read_exact consumes it. + if self.phase == "delete": + self.window.begin_delete(self.deadline) + try: + return super().invoke(operation, **arguments) + finally: + self.window.finish_delete() + started = self.window.begin_request(self.worker) + try: + response = super().invoke(operation, **arguments) + except BaseException: + self.finish_load_request(started, False) + raise + if operation == "get_object": + self.pending_read = started + else: + self.finish_load_request(started, True) + return response + + def finish_load_request(self, started, success): + overlap = self.window.finish_request(self.worker, started, success) + if overlap is not None: + self.emit(kind="load_overlap", worker=self.worker, success=success, **overlap) + + def finish_read(self, success): + if self.pending_read is not None: + started, self.pending_read = self.pending_read, None + self.finish_load_request(started, success) + + def call(self, operation, **arguments): + started = time.monotonic() + code = None + try: + require(started < self.deadline, "PhaseDeadlineExceeded") + return super().call(operation, **arguments) + except RequestFailure as exc: + code = exc.code + raise + finally: + self.emit(kind="operation_time", operation=operation, + seconds=round(time.monotonic() - started, 6), error_code=code) + + +def delete_once(session, bucket, objects): + require(0 < len(objects) <= 1000, "InvalidDeleteBatchSize") + response = session.call("delete_objects", Bucket=bucket, + Delete={"Objects": objects, "Quiet": False}) + deleted, errors = response.get("Deleted", []), response.get("Errors", []) + require(isinstance(deleted, list) and isinstance(errors, list), "InvalidDeleteResponse") + rows = deleted + errors + require(all(isinstance(row, dict) and isinstance(row.get("Key"), str) for row in rows) + and all(isinstance(row.get("Code"), str) for row in errors), "InvalidDeleteResponse") + versioned = all("VersionId" in obj for obj in objects) + identity = lambda row: (row["Key"], row.get("VersionId")) if versioned else row["Key"] + # Count request entries, including duplicate keys; they are not unique objects. + complete = Counter(map(identity, rows)) == Counter(map(identity, objects)) + retained = [{k: row[k] for k in ("Key", "VersionId", "Code") if k in row} for row in errors] + acknowledgments = [{k: row[k] for k in ("Key", "VersionId") if k in row} for row in deleted] + session.emit(kind="delete_response", requested_entries=len(objects), + distinct_keys=len({obj["Key"] for obj in objects}), errors=retained, + deleted=acknowledgments, complete=complete) + require(complete, "IncompleteDeleteResponse") + return retained + + +def listed(session, bucket, versions=False, prefix=""): + result, seen, objects_seen, pagination = [], set(), set(), {} + while True: + operation = "list_object_versions" if versions else "list_objects_v2" + response = session.call(operation, Bucket=bucket, Prefix=prefix, MaxKeys=1000, **pagination) + rows = [] + for field in (("Versions", "DeleteMarkers") if versions else ("Contents",)): + page = response.get(field, []) + require(isinstance(page, list) and all(isinstance(row, dict) + and isinstance(row.get("Key"), str) + and (not versions or isinstance(row.get("VersionId"), str) and row["VersionId"]) + for row in page), "InvalidListResponse") + rows.extend({"Key": row["Key"], **({"VersionId": row["VersionId"]} if versions else {})} + for row in page) + identities = [(row["Key"], row.get("VersionId")) for row in rows] + require(len(set(identities)) == len(identities) and not objects_seen.intersection(identities), + "DuplicateListEntry") + objects_seen.update(identities) + result.extend(rows) + if not response.get("IsTruncated"): + return result + if versions: + marker = response.get("NextKeyMarker") + pagination = {"KeyMarker": marker} + if response.get("NextVersionIdMarker"): + pagination["VersionIdMarker"] = response["NextVersionIdMarker"] + else: + marker = response.get("NextContinuationToken") + pagination = {"ContinuationToken": marker} + token = tuple(sorted(pagination.items())) + require(rows and isinstance(marker, str) and marker and token not in seen, "InvalidListPagination") + seen.add(token) + + +def read_exact(session, bucket, key, payload, version=None): + arguments = {"Bucket": bucket, "Key": key} + if version is not None: + arguments["VersionId"] = version + response = session.call("get_object", **arguments) + body, success = None, False + try: + body = response["Body"] + actual = body.read(len(payload) + 1) + require(actual == payload, "ReadIntegrityMismatch") + success = True + finally: + try: + session.finish_read(success) + finally: + if body is not None: + body.close() + session.emit(kind="read_integrity", key=key, version_id=version, + sha256=hashlib.sha256(actual).hexdigest()) + + +def absent(session, bucket, key): + try: + session.call("head_object", Bucket=bucket, Key=key) + except RequestFailure as exc: + if exc.code in ("404", "NoSuchKey", "NotFound"): + return + raise + raise RequestFailure("DeletedKeyStillReadable") + + +def cleanup_bucket(session, state, result): + receipt = read_state(state) + require(receipt.get("created") is True and receipt.get("endpoint") == session.endpoint, + "UnownedCleanupReceipt") + bucket = receipt["bucket"] + receipt.update(cleanup="FAIL", cleanup_in_progress=True, + cleanup_error_entries=receipt.get("cleanup_error_entries", 0), + cleanup_had_errors=receipt.get("cleanup_had_errors", False)) + save(state, receipt) + try: + abort_uploads(session, bucket) + objects = listed(session, bucket, versions=True) + for offset in range(0, len(objects), 1000): + pending = objects[offset:offset + 1000] + for attempt in range(3): + errors = delete_once(session, bucket, pending) + receipt["cleanup_error_entries"] += len(errors) + receipt["cleanup_had_errors"] |= bool(errors) + save(state, receipt) + if not errors: + break + if any(row["Code"] not in RETRYABLE for row in errors) or attempt == 2: + raise RequestFailure("CleanupDeleteFailed") + pending = [{key: row[key] for key in ("Key", "VersionId")} for row in errors] + else: + raise RequestFailure("CleanupDeleteFailed") + remaining = listed(session, bucket, versions=True) + visible = listed(session, bucket) + receipt.update(remaining_versions=len(remaining), remaining_objects=len(visible)) + require(not remaining and not visible, "CleanupResidue") + session.call("delete_bucket", Bucket=bucket) + receipt["cleanup"] = "PASS" + except Exception as exc: + receipt.update(cleanup_had_errors=True, cleanup_error_code=getattr(exc, "code", type(exc).__name__)) + finally: + receipt["cleanup_in_progress"] = False + save(state, receipt) + result.update({key: receipt[key] for key in ("cleanup", "cleanup_had_errors", "cleanup_error_entries")}) + result["remaining_versions"] = receipt.get("remaining_versions") + result["remaining_objects"] = receipt.get("remaining_objects") + + +class LoadWindow: + """Measure client operation overlap, excluding warmup and scheduling gaps.""" + + def __init__(self, workers): + self.condition = threading.Condition() + self.workers = workers + self.active = set() + self.finished_workers = 0 + self.started = None + self.finished = None + self.results = [{"post_warmup_requests": 0, "overlapping_requests": 0, "overlapping_successes": 0} + for _ in range(workers)] + + def begin_request(self, worker): + with self.condition: + started = time.monotonic() + self.active.add(worker) + self.results[worker]["post_warmup_requests"] += 1 + self.condition.notify_all() + return started + + def finish_request(self, worker, started, success): + finished = time.monotonic() + with self.condition: + self.active.remove(worker) + if self.started is None: + return None + overlap = min(finished, self.finished if self.finished is not None else finished) - max(started, self.started) + if overlap > 0: + span = {"started_monotonic": started, "finished_monotonic": finished, "overlap_seconds": overlap} + self.results[worker]["overlapping_requests"] += 1 + self.results[worker]["overlapping_successes"] += int(success) + self.results[worker].setdefault("first_overlap", {**span, "success": success}) + return span + return None + + def finish_worker(self): + with self.condition: + self.finished_workers += 1 + self.condition.notify_all() + + def begin_delete(self, deadline): + with self.condition: + self.condition.wait_for(lambda: self.active or self.finished_workers == self.workers, + timeout=max(0, deadline - time.monotonic())) + require(self.active and time.monotonic() < deadline, "LoadNotActive") + self.started = time.monotonic() + + def finish_delete(self): + with self.condition: + self.finished = time.monotonic() + + +def load_worker(session, bucket, key, write, stop, ready, release, rounds, window, worker): + result = {"successes": 0, "failures": 0, "first_success": False, + "started_at": now(), "error_codes": []} + try: + for index in range(rounds): + if stop.is_set() or time.monotonic() >= session.deadline: + break + try: + if index: + session.window, session.worker = window, worker + if write: + session.call("put_object", Bucket=bucket, Key=key, Body=PAYLOAD) + else: + read_exact(session, bucket, key, PAYLOAD) + result["successes"] += 1 + result["first_success"] = True + ready.set() + if index == 0: + release.wait(max(0, session.deadline - time.monotonic())) + except Exception as exc: + result["failures"] += 1 + if len(result["error_codes"]) < 5: + result["error_codes"].append(getattr(exc, "code", type(exc).__name__)) + break + finally: + # Covers malformed responses or diagnostics failures before + # read_exact can close a successfully opened response body. + session.finish_read(False) + # Start post-warmup activity immediately after release. A warmup + # success does not establish that any load overlapped the delete. + if index: + stop.wait(0.01) + finally: + result["finished_at"] = now() + result.update(window.results[worker]) + window.finish_worker() + return result + + +def run_case(case_id, mode, count, endpoints, directory, workers=32, factory=None, + prepare_budget=300, request_budget=60, load_budget=30, cleanup_budget=180): + directory.mkdir(parents=True, exist_ok=False) + case_started = time.monotonic() + state, events = directory / "bucket.s3.json", directory / "s3-events.jsonl" + result = {"case_id": case_id, "mode": mode, "batch_size": count, "endpoints": endpoints, + "verdict": "RUNNING", "recovery": "NOT_RUN", "cleanup": "NOT_RUN", "started_at": now(), "phases": []} + save(directory / "result.json", result) + phase, phase_started = "prepare", time.monotonic() + stop, release = threading.Event(), threading.Event() + window = LoadWindow(workers) + futures, pool = [], None + deadlines = {} + def session(endpoint, phase_name, budget): + requests = Requests(endpoint, events, budget, phase_name, factory) + requests.deadline = deadlines.setdefault(phase_name, requests.deadline) + return requests + def transition(name): + nonlocal phase, phase_started + result["phases"].append({"phase": phase, "seconds": round(time.monotonic() - phase_started, 3)}) + phase, phase_started = name, time.monotonic() + def failed(exc, failure_phase): + if failure_phase != "recovery": + result["primary_verdict"] = "FAIL" + result.setdefault("failure_phase", failure_phase) + result.setdefault("error_code", getattr(exc, "code", type(exc).__name__)) + result.setdefault("failures", []).append({"phase": failure_phase, + "error_code": getattr(exc, "code", type(exc).__name__)}) + try: + setup = session(endpoints[0], phase, prepare_budget) + bucket = prepare(setup, state, "s3-bulk-regression")["bucket"] + result["bucket"] = bucket + keys = [f"target/{index:04}" for index in range(count)] + history = {} + if mode != "off": + setup.call("put_bucket_versioning", Bucket=bucket, VersioningConfiguration={"Status": "Enabled"}) + if mode == "suspended": + for key in (keys[0], keys[-1]): + response = setup.call("put_object", Bucket=bucket, Key=key, Body=PAYLOAD) + require(response.get("VersionId") not in (None, "", "null"), "MissingHistoricalVersion") + history[key] = response["VersionId"] + setup.call("put_bucket_versioning", Bucket=bucket, VersioningConfiguration={"Status": "Suspended"}) + for key in keys + [CONTROL]: + response = setup.call("put_object", Bucket=bucket, Key=key, Body=PAYLOAD) + if mode == "on" and key in (keys[0], keys[-1]): + require(response.get("VersionId") not in (None, "", "null"), "MissingHistoricalVersion") + history[key] = response["VersionId"] + for endpoint in endpoints: + seed = session(endpoint, phase, prepare_budget) + require({row["Key"] for row in listed(seed, bucket, prefix="target/")} == set(keys), + "SeedInventoryMismatch") + objects = [{"Key": key} for key in keys] + if case_id in ("S3C-214", "S3C-1307"): + objects = [{"Key": keys[0]}] * 16 + objects[1:] + if case_id == "S3C-214": + objects += [{"Key": "already-absent"}] * 2 + loaded = case_id in ("S3C-1306", "S3C-1307") + if loaded: + transition("load-start") + pool = ThreadPoolExecutor(max_workers=workers) + ready = [] + for index in range(workers): + event = threading.Event() + ready.append(event) + endpoint = endpoints[index % len(endpoints)] + futures.append(pool.submit(load_worker, session(endpoint, "load", load_budget), bucket, + keys[0] if case_id == "S3C-1307" else CONTROL, + case_id == "S3C-1307", stop, event, release, + 32 if case_id == "S3C-1307" else 4096, window, index)) + deadline = time.monotonic() + min(10, load_budget) + for event, future in zip(ready, futures): + while not event.is_set() and not future.done() and time.monotonic() < deadline: + event.wait(min(0.01, max(0, deadline - time.monotonic()))) + require(all(event.is_set() and not future.done() for event, future in zip(ready, futures)), + "LoadNotReady") + transition("delete") + deletion = session(endpoints[0], phase, request_budget) + if loaded: + deletion.window = window + result.update(primary_verdict="PASS", requested_entries=len(objects), distinct_keys=len(keys)) + release.set() + try: + errors = delete_once(deletion, bucket, objects) + result.update(delete_error_entries=len(errors), delete_errors=errors) + if errors: + failed(RequestFailure("DeleteObjectErrors"), phase) + except RequestFailure as exc: + failed(exc, phase) + stop.set() + if pool: + pool.shutdown(wait=True) + pool = None + result["load"] = [future.result() for future in futures] + if loaded: + result["load_overlap"] = { + "delete_started_monotonic": window.started, "delete_finished_monotonic": window.finished, + "successful_requests": sum(row["overlapping_successes"] for row in result["load"]), + "successful_workers": sum(row["overlapping_successes"] > 0 for row in result["load"]), + } + if not result["load_overlap"]["successful_requests"]: + failed(RequestFailure("LoadDidNotOverlapDelete"), "load") + if any(row["failures"] or not row["first_success"] for row in result["load"]): + failed(RequestFailure("LoadRequestFailed"), "load") + transition("verify") + try: + hot_views = [] + for endpoint in endpoints: + check = session(endpoint, phase, request_budget) + remaining = listed(check, bucket, prefix="target/") + if case_id == "S3C-1307": + require(all(row["Key"] == keys[0] for row in remaining), "UnrelatedKeysNotDeleted") + visible = any(row["Key"] == keys[0] for row in remaining) + hot_views.append(visible) + if visible: + read_exact(check, bucket, keys[0], PAYLOAD) + else: + require(not remaining, "DeleteResidue") + samples = keys[1:] if case_id == "S3C-1307" else keys + for key in dict.fromkeys(samples[:1] + samples[-1:]): + absent(check, bucket, key) + read_exact(check, bucket, CONTROL, PAYLOAD) + for key, version in history.items(): + read_exact(check, bucket, key, PAYLOAD, version) + require(len(set(hot_views)) <= 1, "ReplicaViewMismatch") + except Exception as exc: + failed(exc, phase) + if case_id == "S3C-1307": + transition("recovery") + result["recovery"] = "FAIL" + # A fresh same-key PUT/DELETE after all writers join detects leaked ownership. + recover = session(endpoints[0], phase, request_budget) + sentinel = b"after-contention-sentinel" + recover.call("put_object", Bucket=bucket, Key=keys[0], Body=sentinel) + for endpoint in endpoints: + read_exact(session(endpoint, phase, request_budget), bucket, keys[0], sentinel) + require(not delete_once(recover, bucket, [{"Key": keys[0]}]), "RecoveryDeleteFailed") + cold = listed(recover, bucket, prefix="target/") + if cold: + require(not delete_once(recover, bucket, cold), "RecoveryDeleteFailed") + for endpoint in endpoints: + check = session(endpoint, phase, request_budget) + require(not listed(check, bucket, prefix="target/"), "RecoveryResidue") + absent(check, bucket, keys[0]) + result["recovery"] = "PASS" + result["verdict"] = result["primary_verdict"] + except BaseException as exc: + failed(exc, phase) + result["verdict"] = "FAIL" + if isinstance(exc, KeyboardInterrupt): + result["interrupted"] = True + finally: + stop.set() + release.set() + if pool: + pool.shutdown(wait=True) + result["load"] = [future.result() for future in futures] + transition("cleanup") + # Save the primary verdict before recovery can change or fail anything. + save(directory / "result.json", result) + if state.exists(): + try: + cleanup_bucket(session(endpoints[0], phase, cleanup_budget), state, result) + except BaseException as exc: + result.update(cleanup="FAIL", cleanup_error_code=getattr(exc, "code", type(exc).__name__)) + if result["cleanup"] != "PASS" or result.get("cleanup_had_errors"): + result["verdict"] = "FAIL" + result["phases"].append({"phase": phase, "seconds": round(time.monotonic() - phase_started, 3)}) + records = [json.loads(line) for line in events.read_text().splitlines()] if events.exists() else [] + result["http_503_responses"] = sum(row.get("kind") == "s3_request" and row.get("http_status") == 503 + for row in records) + result["service_unavailable_entries"] = sum(row["Code"] == "ServiceUnavailable" + for record in records if record.get("kind") == "delete_response" for row in record["errors"]) + result["phase_seconds"] = {row["phase"]: row["seconds"] for row in result["phases"]} + result["http_503_by_phase"] = dict(Counter(row["phase"] for row in records + if row.get("kind") == "s3_request" and row.get("http_status") == 503)) + result["duration_seconds"] = round(time.monotonic() - case_started, 3) + result["finished_at"] = now() + save(directory / "result.json", result) + return result + + +def plan(groups, sizes): + return [(case_id, mode, count) for case_id in groups for mode in MODES + for count in (sizes if case_id == "S3C-213" else [32] if case_id == "S3C-214" + else [1000] if case_id == "S3C-1306" else [128])] + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--endpoint", action="append", required=True) + parser.add_argument("--result-dir", type=Path, required=True) + parser.add_argument("--case", choices=GROUPS, action="append") + parser.add_argument("--batch-sizes", default="1,32,128,500,1000") + parser.add_argument("--workers", type=int, default=32) + parser.add_argument("--prepare-budget", type=int, default=300) + parser.add_argument("--request-budget", type=int, default=60) + parser.add_argument("--load-budget", type=int, default=30) + parser.add_argument("--cleanup-budget", type=int, default=180) + args = parser.parse_args(argv) + try: + sizes = [int(value) for value in args.batch_sizes.split(",")] + endpoints = [urlsplit(value) for value in args.endpoint] + require(sizes and len(set(sizes)) == len(sizes) and all(1 <= n <= 1000 for n in sizes) + and 1 <= args.workers <= 64 + and all(1 <= value <= 3600 for value in (args.prepare_budget, args.request_budget, + args.load_budget, args.cleanup_budget)) + and len(set(args.endpoint)) == len(args.endpoint) + and all(e.scheme in ("http", "https") and e.hostname and not e.username and not e.password + and e.path in ("", "/") and not e.query and not e.fragment for e in endpoints), "InvalidArguments") + groups = args.case or list(GROUPS) + require(len(set(groups)) == len(groups), "DuplicateCaseSelection") + credentials() + except (ValueError, RequestFailure) as exc: + parser.error(getattr(exc, "code", "InvalidArguments")) + os.umask(0o077) + args.result_dir.mkdir(parents=True, exist_ok=False) + interrupted = False + def terminate(*_): + nonlocal interrupted + if not interrupted: + interrupted = True + raise KeyboardInterrupt() + signal.signal(signal.SIGTERM, terminate) + results, aborted = [], False + inventory = plan(groups, sizes) + for case_id, mode, count in inventory: + name = f"{case_id}-{mode}-{count}" + if aborted: + result = {"case_id": case_id, "mode": mode, "batch_size": count, "verdict": "NOT_RUN"} + else: + result = run_case(case_id, mode, count, args.endpoint, args.result_dir / name, args.workers, + prepare_budget=args.prepare_budget, request_budget=args.request_budget, + load_budget=args.load_budget, cleanup_budget=args.cleanup_budget) + aborted = result["cleanup"] != "PASS" or result.get("interrupted", False) + results.append(result) + print(json.dumps(result), flush=True) + save(args.result_dir / "summary.json", {"results": results, "expected_cases": len(inventory), + "aborted": aborted}) + lines = ["| 用例 | 版本状态 | 批量大小 | 结果 | 失败阶段 | 无负载恢复 | 清理 | HTTP 503 响应 | 逐对象 ServiceUnavailable 条目 |", + "|---|---|---:|---|---|---|---|---:|---:|"] + for row in results: + lines.append("| " + " | ".join(str(row.get(k, "—")) for k in + ("case_id", "mode", "batch_size", "verdict", "failure_phase", "recovery", "cleanup", + "http_503_responses", "service_unavailable_entries")) + " |") + (args.result_dir / "summary.md").write_text("\n".join(lines) + "\n") + return 130 if interrupted else int(aborted or any(row["verdict"] != "PASS" for row in results)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/test_s3_bulk_delete_regressions.py b/test_s3_bulk_delete_regressions.py new file mode 100644 index 0000000..159c9e4 --- /dev/null +++ b/test_s3_bulk_delete_regressions.py @@ -0,0 +1,608 @@ +#!/usr/bin/env python3 +"""Mutation tests for the bulk-delete campaign, without a RustFS environment.""" +from collections import Counter +import io +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import os +from pathlib import Path +import signal +import subprocess +import tempfile +import threading +import time +import unittest +from unittest.mock import Mock, patch +import xml.etree.ElementTree as ET + +from scripts import s3_bulk_delete_regressions as probe + + +class S3Error(Exception): + def __init__(self, code, status=503): + self.response = {"Error": {"Code": code}, "ResponseMetadata": { + "HTTPStatusCode": status, "RequestId": "failed-request", "RetryAttempts": 0}} + + +class VersionedS3: + def __init__(self, mutation=None): + self.mutation = mutation + self.status = "off" + self.versions = {} + self.calls = [] + self.counter = 0 + self.deleted_bucket = False + self.cleanup_failures = 0 + self.after_delete = False + self.lock = threading.RLock() + self.load_counts = Counter() + self.bulk_entered = threading.Event() + self.load_active = 0 + self.observed_overlap = False + + def load_request(self): + if threading.current_thread() is threading.main_thread(): + return + with self.lock: + worker = threading.get_ident() + self.load_counts[worker] += 1 + warmed = self.load_counts[worker] > 1 + # A successful fixture supplies real post-warmup operations held open + # until delete_objects enters, rather than relying on scheduler luck. + if warmed: + with self.lock: + self.load_active += 1 + try: + if not self.bulk_entered.wait(2): + raise S3Error("FixtureDeleteNotStarted", 500) + finally: + with self.lock: + self.load_active -= 1 + + def append(self, key, body, marker=False): + self.counter += 1 + version = f"v{self.counter}" if self.status == "on" else "null" + rows = self.versions.setdefault(key, []) + if version == "null": + rows[:] = [row for row in rows if row[0] != "null"] + rows.append((version, body, marker)) + return version + + def create_bucket(self, **kwargs): + self.calls.append(("create", kwargs)) + return {} + + def put_bucket_versioning(self, **kwargs): + self.status = "on" if kwargs["VersioningConfiguration"]["Status"] == "Enabled" else "suspended" + self.calls.append(("versioning", self.status)) + return {} + + def put_object(self, **kwargs): + self.load_request() + with self.lock: + self.calls.append(("put", kwargs["Key"])) + version = self.append(kwargs["Key"], kwargs["Body"]) + if self.mutation == "lost-seed" and kwargs["Key"] == "target/0000": + self.versions[kwargs["Key"]] = [] + return {} if self.status == "off" else {"VersionId": version} + + def get_object(self, **kwargs): + self.load_request() + with self.lock: + rows = self.versions.get(kwargs["Key"], []) + if "VersionId" in kwargs: + rows = [row for row in rows if row[0] == kwargs["VersionId"]] + if not rows or rows[-1][2]: + raise S3Error("NoSuchKey", 404) + body = rows[-1][1] + if self.mutation == "corrupt" and kwargs["Key"] == probe.CONTROL: + body = b"torn-payload" + if self.after_delete and kwargs["Key"] == probe.CONTROL and self.mutation in ("read-timeout", "incomplete-read"): + from botocore.exceptions import IncompleteReadError, ReadTimeoutError + error = (ReadTimeoutError(endpoint_url="http://node.invalid") if self.mutation == "read-timeout" + else IncompleteReadError(actual_bytes=1, expected_bytes=len(body))) + self.mutation = None + stream = io.BytesIO(body) + stream.read = Mock(side_effect=error) + return {"Body": stream} + return {"Body": io.BytesIO(body)} + + def head_object(self, **kwargs): + with self.lock: + rows = self.versions.get(kwargs["Key"], []) + if not rows or rows[-1][2]: + raise S3Error("404", 404) + return {"ContentLength": len(rows[-1][1])} + + def list_objects_v2(self, **kwargs): + with self.lock: + rows = [{"Key": key} for key, versions in sorted(self.versions.items()) + if key.startswith(kwargs.get("Prefix", "")) and versions and not versions[-1][2]] + if self.mutation == "hidden-residue" and self.after_delete and kwargs.get("Prefix") == "target/": + rows = [] + return {"Contents": rows, "IsTruncated": False} + + def list_object_versions(self, **kwargs): + with self.lock: + result = {"Versions": [], "DeleteMarkers": [], "IsTruncated": False} + for key, rows in self.versions.items(): + if key.startswith(kwargs.get("Prefix", "")): + for version, _, marker in rows: + result["DeleteMarkers" if marker else "Versions"].append({"Key": key, "VersionId": version}) + return result + + def delete_objects(self, **kwargs): + if "VersionId" not in kwargs["Delete"]["Objects"][0]: + with self.lock: + self.observed_overlap |= self.load_active > 0 + self.bulk_entered.set() + with self.lock: + objects = kwargs["Delete"]["Objects"] + self.calls.append(("delete", objects)) + explicit = "VersionId" in objects[0] + self.after_delete |= not explicit + if not explicit and self.mutation == "http503": + self.mutation = None + raise S3Error("ServiceUnavailable") + if not explicit and self.mutation == "interrupt": + raise KeyboardInterrupt() + failed = self.mutation in ("object-error", "incomplete-object-error") and not explicit + if explicit and self.mutation in ("cleanup-retry", "cleanup-blocked"): + failed = self.mutation == "cleanup-blocked" or self.cleanup_failures == 0 + self.cleanup_failures += 1 + response = {"Deleted": [], "Errors": [], "ResponseMetadata": { + "HTTPStatusCode": 200, "RequestId": "bulk-request", "RetryAttempts": 0}} + for index, obj in enumerate(objects): + key = obj["Key"] + if failed and (explicit or index == 0): + code = "AccessDenied" if self.mutation == "cleanup-blocked" else "ServiceUnavailable" + response["Errors"].append({**obj, "Code": code}) + continue + if explicit: + self.versions[key] = [row for row in self.versions.get(key, []) if row[0] != obj["VersionId"]] + elif self.mutation in ("residue", "hidden-residue") and index == 0: + pass + elif self.status == "off" or self.mutation == "lost-history": + self.versions[key] = [] + else: + self.append(key, b"", marker=True) + response["Deleted"].append(dict(obj)) + if not explicit and self.mutation in ("missing-ack", "incomplete-object-error"): + response["Deleted"].pop() + return response + + def list_multipart_uploads(self, **kwargs): + return {"Uploads": []} + + def delete_bucket(self, **kwargs): + assert not any(self.versions.values()) + self.deleted_bucket = True + return {} + + +class BulkDeleteTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + + def run_probe(self, case_id="S3C-213", mode="off", mutation=None, count=3): + fake = VersionedS3(mutation) + directory = self.root / f"{case_id}-{mode}-{mutation}" + result = probe.run_case(case_id, mode, count, ["http://node1.invalid", "http://node2.invalid"], + directory, workers=2, factory=lambda _: fake) + events = [json.loads(line) for line in (directory / "s3-events.jsonl").read_text().splitlines()] + return result, fake, events + + def test_all_groups_and_version_states_execute_and_remove_versions_and_markers(self): + for case_id in probe.GROUPS: + for mode in probe.MODES: + with self.subTest(case_id=case_id, mode=mode): + result, fake, events = self.run_probe(case_id, mode) + self.assertEqual(result["verdict"], "PASS", result) + self.assertEqual(result["cleanup"], "PASS") + self.assertEqual(result["remaining_versions"], 0) + self.assertTrue(fake.deleted_bucket) + self.assertTrue(any(row.get("kind") == "operation_time" and row["phase"] == "delete" for row in events)) + if case_id == "S3C-1307": + self.assertEqual(result["recovery"], "PASS") + if case_id in ("S3C-1306", "S3C-1307"): + self.assertTrue(fake.observed_overlap) + self.assertGreater(result["load_overlap"]["successful_requests"], 0) + overlapping = [row for row in events if row.get("kind") == "load_overlap" and row["success"]] + self.assertTrue(overlapping) + interval = result["load_overlap"] + for row in overlapping: + self.assertLess(row["started_monotonic"], interval["delete_finished_monotonic"]) + self.assertGreater(row["finished_monotonic"], interval["delete_started_monotonic"]) + self.assertGreater(row["overlap_seconds"], 0) + if case_id == "S3C-214": + batches = [row for operation, row in fake.calls if operation == "delete"] + self.assertEqual(Counter(row["Key"] for row in batches[0])["target/0000"], 16) + self.assertEqual(result["requested_entries"], 20) + + def test_http200_object_error_is_a_failure_even_when_cleanup_succeeds(self): + result, fake, _ = self.run_probe(mutation="object-error") + self.assertEqual(result["verdict"], "FAIL") + self.assertEqual(result["http_503_responses"], 0) + self.assertEqual(result["service_unavailable_entries"], 1) + self.assertEqual(result["delete_error_entries"], 1) + self.assertEqual(result["failure_phase"], "delete") + self.assertEqual(result["cleanup"], "PASS") + self.assertTrue(fake.deleted_bucket) + + def test_http503_is_one_response_and_is_not_expanded_into_object_errors(self): + result, _, events = self.run_probe(mutation="http503") + self.assertEqual(result["verdict"], "FAIL") + self.assertEqual(result["failure_phase"], "delete") + self.assertEqual(result["http_503_responses"], 1) + self.assertEqual(result["service_unavailable_entries"], 0) + response = next(row for row in events if row.get("http_status") == 503) + self.assertEqual(response["request_id"], "failed-request") + self.assertEqual(response["sdk_retries"], 0) + self.assertEqual(result["cleanup"], "PASS") + + def test_incomplete_mixed_response_keeps_per_object_error_evidence(self): + result, _, events = self.run_probe(mutation="incomplete-object-error") + self.assertEqual(result["verdict"], "FAIL") + self.assertEqual(result["error_code"], "IncompleteDeleteResponse") + self.assertEqual(result["service_unavailable_entries"], 1) + self.assertEqual(result["http_503_responses"], 0) + response = next(row for row in events if row.get("kind") == "delete_response" and row["phase"] == "delete") + self.assertEqual(response["errors"], [{"Key": "target/0000", "Code": "ServiceUnavailable"}]) + self.assertEqual(response["deleted"], [{"Key": "target/0001"}]) + self.assertFalse(response["complete"]) + self.assertEqual(result["cleanup"], "PASS") + + def test_load_client_is_reused_until_its_read_timeout_exceeds_remaining_budget(self): + factory = Mock(side_effect=lambda remaining: Mock(put_object=Mock(return_value={}))) + with patch.object(probe.time, "monotonic", return_value=0) as clock: + requests = probe.Requests("http://node.invalid", self.root / "reuse-events", 30, "load", factory) + for instant in (0, 1, 19, 20): + clock.return_value = instant + requests.call("put_object", Bucket="owned", Key="hot", Body=b"payload") + self.assertEqual(factory.call_count, 1) + original = requests.s3 + clock.return_value = 21 + requests.call("put_object", Bucket="owned", Key="hot", Body=b"payload") + self.assertIsNot(requests.s3, original) + self.assertEqual(factory.call_args.args, (9,)) + + def test_loaded_case_with_only_warmup_cannot_pass(self): + worker = probe.load_worker + def warmup_only(*args): + arguments = list(args) + arguments[7] = 1 + return worker(*arguments) + for case_id in ("S3C-1306", "S3C-1307"): + with self.subTest(case_id=case_id), patch.object(probe, "load_worker", warmup_only): + result, fake, _ = self.run_probe(case_id=case_id) + self.assertEqual(result["verdict"], "FAIL") + self.assertEqual(result["error_code"], "LoadNotActive") + self.assertFalse(fake.observed_overlap) + self.assertEqual(result["load_overlap"]["successful_requests"], 0) + self.assertTrue(all(row["successes"] == 1 and row["post_warmup_requests"] == 0 for row in result["load"])) + self.assertEqual(result["cleanup"], "PASS") + + def test_completed_post_warmup_requests_cannot_substitute_for_overlap(self): + # Reproduce a scheduling gap before the actual delete invocation: + # every load request has already returned when deletion begins. + original = probe.LoadWindow.begin_delete + def after_workers(window, deadline): + with window.condition: + finished = window.condition.wait_for(lambda: window.finished_workers == window.workers, timeout=2) + self.assertTrue(finished) + original(window, deadline) + worker = probe.load_worker + def two_rounds(*args): + arguments = list(args) + arguments[7] = 2 + return worker(*arguments) + for case_id in ("S3C-1306", "S3C-1307"): + with self.subTest(case_id=case_id), patch.object(probe, "load_worker", two_rounds), \ + patch.object(probe.LoadWindow, "begin_delete", after_workers), \ + patch.object(VersionedS3, "load_request", lambda _self: None): + result, _, _ = self.run_probe(case_id=case_id) + self.assertEqual(result["verdict"], "FAIL") + self.assertEqual(result["load_overlap"]["successful_requests"], 0) + self.assertTrue(all(row["post_warmup_requests"] == 1 for row in result["load"])) + self.assertEqual(result["cleanup"], "PASS") + + def test_recorded_overlap_uses_intersection_not_total_request_duration(self): + for started, finished, expected in ((1, 4, 2), (3, 6, 2), (0, 1, 0), (5, 7, 0)): + with self.subTest(started=started, finished=finished): + window = probe.LoadWindow(1) + window.started, window.finished = 2, 5 + with patch.object(probe.time, "monotonic", return_value=started): + entered = window.begin_request(0) + with patch.object(probe.time, "monotonic", return_value=finished): + overlap = window.finish_request(0, entered, True) + self.assertEqual(overlap["overlap_seconds"] if overlap else 0, expected) + + def test_stream_remains_active_until_body_is_consumed_and_closed(self): + fake = VersionedS3() + fake.put_object(Key=probe.CONTROL, Body=probe.PAYLOAD) + window = probe.LoadWindow(1) + requests = probe.Requests("http://node.invalid", self.root / "stream-events", 30, "load", lambda _: fake) + requests.window, requests.worker = window, 0 + body = io.BytesIO(probe.PAYLOAD) + read = body.read + def during_read(size): + self.assertIn(0, window.active) + window.begin_delete(time.monotonic() + 1) + window.finish_delete() + return read(size) + body.read = during_read + with patch.object(fake, "get_object", return_value={"Body": body}): + probe.read_exact(requests, "owned", probe.CONTROL, probe.PAYLOAD) + self.assertTrue(body.closed) + self.assertFalse(window.active) + self.assertEqual(window.results[0]["overlapping_successes"], 1) + + def test_overlap_diagnostic_failure_still_closes_the_stream(self): + body = io.BytesIO(probe.PAYLOAD) + requests = probe.Requests("http://node.invalid", self.root / "close-events", 30, "load") + with patch.object(requests, "call", return_value={"Body": body}), \ + patch.object(requests, "finish_read", side_effect=OSError("fixture evidence failure")): + with self.assertRaises(OSError): + probe.read_exact(requests, "owned", probe.CONTROL, probe.PAYLOAD) + self.assertTrue(body.closed) + + def test_recovery_failure_does_not_change_a_passed_primary_verdict(self): + original = probe.Requests.call + def fail_recovery(requests, operation, **arguments): + if requests.phase == "recovery" and operation == "put_object": + raise probe.RequestFailure("RecoveryFixtureFailure") + return original(requests, operation, **arguments) + with patch.object(probe.Requests, "call", fail_recovery): + result, _, _ = self.run_probe(case_id="S3C-1307") + self.assertEqual(result["primary_verdict"], "PASS") + self.assertEqual(result["recovery"], "FAIL") + self.assertEqual(result["verdict"], "FAIL") + self.assertEqual(result["failure_phase"], "recovery") + self.assertEqual(result["cleanup"], "PASS") + + def test_streamed_verification_failures_still_run_same_key_recovery(self): + for mutation, code in (("read-timeout", "ReadTimeoutError"), ("incomplete-read", "IncompleteReadError")): + with self.subTest(mutation=mutation): + result, _, events = self.run_probe(case_id="S3C-1307", mutation=mutation) + self.assertEqual(result["failure_phase"], "verify") + self.assertEqual(result["error_code"], code) + self.assertEqual(result["primary_verdict"], "FAIL") + self.assertEqual(result["verdict"], "FAIL") + self.assertEqual(result["recovery"], "PASS") + self.assertTrue(any(row.get("kind") == "s3_request" and row["phase"] == "recovery" for row in events)) + self.assertEqual(result["cleanup"], "PASS") + + def test_real_s3_responses_keep_entry_counts_and_disable_transport_retries(self): + calls = [] + class Handler(BaseHTTPRequestHandler): + def log_message(self, *_): + pass + + def do_POST(self): + body = ET.fromstring(self.rfile.read(int(self.headers["Content-Length"]))) + keys = [row.text for row in body.iter() if row.tag.rsplit("}", 1)[-1] == "Key"] + calls.append(keys) + if server.top_level_failure: + result = ET.Element("Error") + ET.SubElement(result, "Code").text = "ServiceUnavailable" + else: + result = ET.Element("DeleteResult", xmlns="http://s3.amazonaws.com/doc/2006-03-01/") + for index, key in enumerate(keys): + row = ET.SubElement(result, "Error" if index == 0 else "Deleted") + ET.SubElement(row, "Key").text = key + if index == 0: + ET.SubElement(row, "Code").text = "ServiceUnavailable" + response = ET.tostring(result) + self.send_response(503 if server.top_level_failure else 200) + self.send_header("Content-Type", "application/xml") + self.send_header("x-amz-request-id", "real-http200") + self.send_header("Content-Length", str(len(response))) + self.end_headers() + self.wfile.write(response) + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + server.top_level_failure = False + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + with patch.dict(os.environ, {"RUSTFS_ACCESS_KEY": "test-access", "RUSTFS_SECRET_KEY": "test-secret", + "RUSTFS_SESSION_TOKEN": ""}): + session = probe.Requests(f"http://127.0.0.1:{server.server_port}", self.root / "http-events", 30, "delete") + errors = probe.delete_once(session, "owned-bucket", [{"Key": "duplicate"}, + {"Key": "duplicate"}, {"Key": "cold"}]) + server.top_level_failure = True + failed_session = probe.Requests(f"http://127.0.0.1:{server.server_port}", self.root / "http503-events", 30, "delete") + with self.assertRaises(probe.RequestFailure) as failure: + probe.delete_once(failed_session, "owned-bucket", [{"Key": "cold"}]) + self.assertEqual(failure.exception.code, "ServiceUnavailable") + self.assertEqual(failed_session.http_503, 1) + self.assertEqual(errors, [{"Key": "duplicate", "Code": "ServiceUnavailable"}]) + self.assertEqual(calls, [["duplicate", "duplicate", "cold"], ["cold"]]) + self.assertEqual(session.http_503, 0) + records = [json.loads(line) for line in (self.root / "http-events").read_text().splitlines()] + request = next(row for row in records if row["kind"] == "s3_request") + self.assertEqual(request["request_id"], "real-http200") + self.assertEqual(request["sdk_retries"], 0) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + + def test_missing_acknowledgment_residue_corruption_and_lost_history_fail(self): + for mutation in ("missing-ack", "residue", "hidden-residue", "corrupt", "lost-history"): + with self.subTest(mutation=mutation): + result, _, _ = self.run_probe(mode="on", mutation=mutation) + self.assertEqual(result["verdict"], "FAIL") + self.assertEqual(result["cleanup"], "PASS") + + def test_verify_budget_is_shared_across_endpoints(self): + clock = [0.0] + original = probe.Requests.call + def consuming_call(requests, operation, **arguments): + response = original(requests, operation, **arguments) + if requests.phase == "verify" and requests.endpoint == "http://node1.invalid" and operation == "get_object": + clock[0] = 61.0 + return response + with patch.object(probe.time, "monotonic", side_effect=lambda: clock[0]), \ + patch.object(probe.Requests, "call", consuming_call): + result, _, events = self.run_probe() + self.assertEqual(result["verdict"], "FAIL") + self.assertEqual(result["failure_phase"], "verify") + self.assertEqual(result["error_code"], "PhaseDeadlineExceeded") + self.assertFalse(any(row.get("kind") == "s3_request" and row.get("phase") == "verify" + and row.get("endpoint") == "http://node2.invalid" for row in events)) + self.assertEqual(result["cleanup"], "PASS") + + def test_cleanup_retry_preserves_first_error_and_cannot_turn_case_green(self): + result, _, _ = self.run_probe(mutation="cleanup-retry") + self.assertEqual(result["primary_verdict"], "PASS") + self.assertEqual(result["cleanup"], "PASS") + self.assertTrue(result["cleanup_had_errors"]) + self.assertEqual(result["verdict"], "FAIL") + self.assertGreater(result["cleanup_error_entries"], 0) + + def test_unrecoverable_cleanup_keeps_bucket_and_fails(self): + result, fake, _ = self.run_probe(mutation="cleanup-blocked") + self.assertEqual(result["cleanup"], "FAIL") + self.assertEqual(result["verdict"], "FAIL") + self.assertFalse(fake.deleted_bucket) + + def test_interrupt_saves_primary_failure_and_runs_cleanup(self): + result, fake, _ = self.run_probe(mutation="interrupt") + self.assertTrue(result["interrupted"]) + self.assertEqual(result["failure_phase"], "delete") + self.assertEqual(result["verdict"], "FAIL") + self.assertTrue(fake.deleted_bucket) + + def test_failed_get_load_is_not_mistaken_for_successful_warmup(self): + result, fake, _ = self.run_probe(case_id="S3C-1306", mutation="corrupt") + self.assertEqual(result["failure_phase"], "load-start") + self.assertEqual(result["error_code"], "LoadNotReady") + self.assertEqual(result["verdict"], "FAIL") + self.assertFalse(any(operation == "delete" and "VersionId" not in rows[0] + for operation, rows in fake.calls if operation == "delete")) + + def test_expired_phase_never_dispatches_a_request(self): + fake = VersionedS3() + requests = probe.Requests("http://node.invalid", self.root / "events", -1, "delete", lambda _: fake) + with self.assertRaises(probe.RequestFailure) as error: + requests.call("put_object", Bucket="owned", Key="key", Body=b"data") + self.assertEqual(error.exception.code, "PhaseDeadlineExceeded") + self.assertEqual(fake.calls, []) + + def test_missing_seed_cannot_make_an_idempotent_delete_look_like_a_regression_pass(self): + result, fake, _ = self.run_probe(mutation="lost-seed") + self.assertEqual(result["verdict"], "FAIL") + self.assertEqual(result["failure_phase"], "prepare") + self.assertEqual(result["error_code"], "SeedInventoryMismatch") + self.assertFalse(fake.after_delete) + + def test_empty_or_repeated_pagination_is_rejected(self): + for rows in ([], [{"Key": "same", "VersionId": "v1"}]): + requests = probe.Requests("http://node.invalid", self.root / "events", 30, "cleanup") + with patch.object(requests, "call", return_value={"Versions": rows, "IsTruncated": True, + "NextKeyMarker": "same", "NextVersionIdMarker": "v1"}): + with self.assertRaises(probe.RequestFailure): + probe.listed(requests, "owned", versions=True) + + def test_unowned_receipt_cannot_trigger_cleanup(self): + fake = VersionedS3() + state = self.root / "state.json" + state.write_text(json.dumps({"created": False, "bucket": "other", "endpoint": "http://node.invalid"})) + requests = probe.Requests("http://node.invalid", self.root / "events", 30, "cleanup", lambda _: fake) + with self.assertRaises((probe.RequestFailure, ValueError)): + probe.cleanup_bucket(requests, state, {}) + self.assertEqual(fake.calls, []) + + def test_failed_hotspot_delete_still_checks_recovery_and_preserves_failure(self): + result, fake, _ = self.run_probe(case_id="S3C-1307", mode="on", mutation="http503") + self.assertEqual(result["failure_phase"], "delete") + self.assertEqual(result["verdict"], "FAIL") + self.assertEqual(result["recovery"], "PASS") + self.assertTrue(fake.deleted_bucket) + + def test_resumed_cleanup_keeps_previous_failure_history(self): + fake = VersionedS3() + requests = probe.Requests("http://node.invalid", self.root / "events", 30, "cleanup", lambda _: fake) + state = self.root / "state.json" + receipt = probe.prepare(requests, state, "test") + receipt.update(cleanup="FAIL", cleanup_had_errors=True, cleanup_error_entries=7) + probe.save(state, receipt) + result = {} + probe.cleanup_bucket(requests, state, result) + self.assertEqual(result["cleanup"], "PASS") + self.assertTrue(result["cleanup_had_errors"]) + self.assertEqual(result["cleanup_error_entries"], 7) + + def test_cli_records_every_unrun_case_after_failed_cleanup(self): + old_handler = signal.getsignal(signal.SIGTERM) + self.addCleanup(signal.signal, signal.SIGTERM, old_handler) + directory = self.root / "cli" + with patch.dict(os.environ, {"RUSTFS_ACCESS_KEY": "test-access", "RUSTFS_SECRET_KEY": "test-secret"}), \ + patch.object(probe, "run_case", return_value={"verdict": "FAIL", "cleanup": "FAIL"}) as runner: + self.assertEqual(probe.main(["--endpoint", "http://node.invalid", "--result-dir", str(directory)]), 1) + document = json.loads((directory / "summary.json").read_text()) + self.assertEqual(document["expected_cases"], 24) + self.assertEqual(len(document["results"]), 24) + self.assertEqual(sum(row["verdict"] == "NOT_RUN" for row in document["results"]), 23) + self.assertEqual(runner.call_count, 1) + self.assertIn("NOT_RUN", (directory / "summary.md").read_text()) + + def test_invalid_options_do_not_create_artifacts_or_contact_a_server(self): + invalid = [["--endpoint", "http://user:secret@node.invalid"], + ["--endpoint", "ftp://node.invalid"], ["--endpoint", "http://node.invalid", "--workers", "0"], + ["--endpoint", "http://node.invalid", "--batch-sizes", "1001"]] + with patch.object(probe, "run_case") as runner: + for options in invalid: + with self.subTest(options=options), self.assertRaises(SystemExit) as error: + probe.main(options + ["--result-dir", str(self.root / "invalid")]) + self.assertEqual(error.exception.code, 2) + runner.assert_not_called() + self.assertFalse((self.root / "invalid").exists()) + + def test_explicit_credentials_do_not_mix_with_ambient_aws_credentials_or_tokens(self): + with patch.dict(os.environ, {"RUSTFS_ACCESS_KEY": "explicit", "RUSTFS_SECRET_KEY": "explicit-secret", + "AWS_SESSION_TOKEN": "ambient-token"}, clear=True): + self.assertEqual(probe.credentials(), ("explicit", "explicit-secret", None)) + with patch.dict(os.environ, {"RUSTFS_ACCESS_KEY": "partial", "AWS_ACCESS_KEY_ID": "ambient", + "AWS_SECRET_ACCESS_KEY": "ambient-secret"}, clear=True): + with self.assertRaises(probe.RequestFailure): + probe.credentials() + + def test_full_s3_suite_selects_new_cases_without_deploying_in_dry_run(self): + script = Path(__file__).with_name("rustfs-s3-compat-test.sh") + result = subprocess.run(["bash", str(script), "--topology", "single-single", "--dry-run", "-y", + "--cases", ",".join(probe.GROUPS)], + capture_output=True, text=True, timeout=10) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + for case_id in probe.GROUPS: + self.assertIn(f"[DRY-RUN] {case_id} ", result.stdout) + + def test_topology_wrapper_uses_configured_nodes_and_keeps_credentials_out_of_arguments(self): + script = Path(__file__).with_name("rustfs-s3-compat-test.sh") + command = r''' +source "$RUNNER" +ENDPOINT=http://primary.invalid:9000 +NODES=(configured-a configured-b configured-c configured-d) +CURRENT_TOPOLOGY=multi-multi +S3_BULK_RESULTS_DIR="$RESULT_ROOT" +ACCESS_KEY=test-access +SECRET_KEY=must-not-appear-in-arguments +python3() { printf '%s\n' "$@" > "$CALL_LOG"; } +test_s3c213 +''' + result = subprocess.run(["bash", "-c", command], capture_output=True, text=True, timeout=10, + env={**os.environ, "RUNNER": str(script), "RESULT_ROOT": str(self.root / "wrapper"), + "CALL_LOG": str(self.root / "call-log")}) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + arguments = (self.root / "call-log").read_text() + self.assertIn("http://primary.invalid:9000", arguments) + for node in ("configured-b", "configured-c", "configured-d"): + self.assertIn(f"http://{node}:9000", arguments) + self.assertNotIn("must-not-appear-in-arguments", arguments) + + +if __name__ == "__main__": + unittest.main()