fix(testing): guard staging volumes and interrupted capture

Verify volume ownership before privileged staging and disable implicit image-content copying. Preserve failed evidence and continue owned finalization when capture or cleanup is interrupted.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

.config/make/tests.mak

@@ -71,6 +71,7 @@ script-tests: ## Run shell script tests
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_nightly_candidate.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_functional_chain.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_functional_chain_health.py
+	$(RUSTFS_PYTHON_BIN) ./scripts/test_repro_rejoin_readiness_docker.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_validation_registry.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_inspect_validation_runner.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_retain_validation_evidence.py

.docker/rejoin-readiness.Dockerfile

@@ -0,0 +1,40 @@
+# syntax=docker/dockerfile:1.6
+# Copyright 2026 RustFS Team
+# Licensed under the Apache License, Version 2.0.
+
+# A compatible native runtime with Python 3.11+, getent and stat is required by
+# the isolated fixture's staging/provenance checks. This is a test image.
+ARG RUNTIME_BASE=python:3.13-slim-trixie
+FROM ${RUNTIME_BASE}
+
+ARG SOURCE_REVISION
+ARG BINARY_SHA256
+ARG BUILD_PROFILE=dev
+COPY rustfs /usr/bin/rustfs
+RUN python3 - "$SOURCE_REVISION" "$BINARY_SHA256" "$BUILD_PROFILE" <<'PY'
+import hashlib
+import os
+import re
+import shutil
+import sys
+
+source, digest, profile = sys.argv[1:]
+if not re.fullmatch(r"[0-9a-f]{40}", source):
+    raise SystemExit("invalid source revision")
+if not re.fullmatch(r"[0-9a-f]{64}", digest):
+    raise SystemExit("invalid binary digest")
+if profile not in ("dev", "release"):
+    raise SystemExit("invalid build profile")
+with open("/usr/bin/rustfs", "rb") as binary:
+    if hashlib.file_digest(binary, "sha256").hexdigest() != digest:
+        raise SystemExit("binary digest mismatch")
+for tool in ("getent", "stat"):
+    if shutil.which(tool) is None:
+        raise SystemExit(f"missing runtime tool: {tool}")
+os.chmod("/usr/bin/rustfs", 0o755)
+PY
+LABEL io.rustfs.repro.source_revision="${SOURCE_REVISION}" \
+      io.rustfs.repro.binary_sha256="${BINARY_SHA256}" \
+      io.rustfs.repro.build_profile="${BUILD_PROFILE}"
+USER 10001:10001
+ENTRYPOINT ["/usr/bin/rustfs"]

docs/testing/README.md

@@ -41,6 +41,8 @@ The [immutable validation candidate reference](validation-candidates.md) describ
 
 The [distributed multipart RPC delay gates](multipart-rpc-delay.md) define authenticated method-entry barriers, latency and peer-loss oracles, and the boundary between hook evidence and production incident attribution.
 
+The [Docker rejoin readiness reproduction](docker-rejoin-readiness.md) compares node and cluster admission gates with fixed-source HTTPS containers and preserves the first object failure across recovery.
+
 ## Naming conventions
 
 ### Reserved test-name substrings (migration gate)

docs/testing/docker-rejoin-readiness.md

@@ -0,0 +1,54 @@
+# Docker rejoin readiness reproduction
+
+**Use this when:** distinguishing restarted-node admission failures from acknowledged-overwrite peer GET failures in an HTTPS erasure set, or validating a change against fixed-source Docker binaries.
+
+**Source of truth:** `scripts/repro_rejoin_readiness_docker.py` (`Experiment`, `GATE_PATHS`, `cleanup_resources`), `scripts/test_repro_rejoin_readiness_docker.py`, and `.docker/rejoin-readiness.Dockerfile`.
+
+## Scope and prerequisites
+
+The harness operates one isolated project on the selected Docker daemon. It records the client host, daemon architecture, kernel/storage driver, runtime distribution, data-volume filesystem, resource limits and binary identity. Matching Linux AMD64 and ARM64 daemon/image/binary combinations are accepted; a mismatch is refused before server startup. A successful run on a virtualized ARM64 Docker host is functional evidence for that environment, not native AMD64, independent-host durability, or production throughput evidence.
+
+Use Docker, host curl with built-in `--aws-sigv4`, OpenSSL and Python. Run the client on the daemon host or a local Docker VM that forwards published loopback ports to it; selecting a remote Docker context alone does not forward those ports. The runtime image must provide `/bin/sh`, Python 3.11+, `getent`, `stat`, and a compatible `/usr/bin/rustfs`. The fixture keeps the root filesystem read-only, mounts owned Linux data/configuration volumes, verifies TLS, and publishes only loopback S3 ports. It explicitly creates and records a non-internal bridge network; include this difference when comparing it with an internal-network fixture. It does not use production credentials or data. See [probe semantics](../architecture/readiness-matrix.md#probe-semantics) and [recovered-backend routing](../operations/rolling-restart.md#returning-a-recovered-node-to-a-load-balancer).
+
+## Prepare an immutable image
+
+Build the selected source in an isolated checkout with a recorded compiler, target, features and profile. Reusing dependency caches is safe only when package identities and inputs remain correct; use a separate target directory for each source revision. If seeding it from a compiled cache, run `cargo clean --workspace` in the selected source before building; cached local path-package APIs must not cross revisions. Distinct container source paths alone do not establish freshness when checkout mtimes predate cached outputs. Do not label a dirty runtime build as an unmodified commit. Copy the completed binary into a private image context named `rustfs`, and record its actual SHA256. Use a libc-compatible native base with no inherited `RUSTFS_*` or `RUST_LOG` configuration; preflight rejects those overrides so each recorded control uses the fixture's explicit environment. The wrapper's default is an example GNU runtime, not a production-image recommendation.
+
+```bash
+REJOIN_SOURCE_SHA="$(git rev-parse HEAD)"
+REJOIN_BINARY_SHA="$(openssl dgst -sha256 /path/to/context/rustfs | awk '{print $NF}')"
+docker build --pull=false -f .docker/rejoin-readiness.Dockerfile \
+  --build-arg RUNTIME_BASE='<compatible-native-base>@sha256:<digest>' \
+  --build-arg SOURCE_REVISION="$REJOIN_SOURCE_SHA" \
+  --build-arg BINARY_SHA256="$REJOIN_BINARY_SHA" \
+  --build-arg BUILD_PROFILE=release \
+  -t rustfs-rejoin:fixed /path/to/context
+```
+
+The wrapper validates its source/profile/digest arguments. The runner resolves the image to its immutable ID, validates labels, and independently verifies the binary digest and ELF architecture in the isolated staging container before executing it. Its evidence manifest records what was tested; an image name or `--version` string alone is insufficient provenance. An existing historical reproducer's revision checks must be updated together with its new binary and image labels; preserve the original fixture as the control.
+
+## Run controlled cases
+
+```bash
+python3 scripts/repro_rejoin_readiness_docker.py run \
+  --image rustfs-rejoin:fixed --expected-source "$REJOIN_SOURCE_SHA" \
+  --build-profile release --out-dir /path/to/new-private-evidence \
+  --gate node-write --payload-mode distinct --cleanup-on-success
+```
+
+Use a fresh output directory and project for each case. `--gate cluster` preserves the cluster-only recovery condition; `--gate node-write` additionally requires node readiness. Compare the gates on the same binary before attributing a change to code. Compare source revisions with the same architecture, runtime, profile, quotas, payload mode and workload. The default outage duration matches the historical reduced fixture; it is a fault interval, not an extra delay after the recovery gate. `--workers 2` is a separate worker-count experiment and must not silently change the default control.
+
+`--payload-mode original` preserves identical overwrite bytes. The distinct mode retains the last acknowledged payload per key so an old matching version cannot masquerade as the latest overwrite. `--stress-iterations` increases same-key overwrite/peer-read pairs during each outage. Bucket creation may retry startup 503; object operations have no automatic retries. A first failed object response remains a failed result even if recovery or later observations succeed. Cluster-only health may pass before S3 admission; that observation is distinct from a survivor's namespace-lock timeout.
+
+## Failure evidence and cleanup
+
+The runner saves request timestamps, status, response bodies/headers, acknowledged-content identities, container generations, phase snapshots, Docker output and runtime logs. On a namespace-lock timeout it samples the read-only signed `/rustfs/admin/v3/top/locks` endpoint on the surviving nodes before restarting stopped peers. A missing/unsupported diagnostic endpoint is recorded, not interpreted as an empty lock table. No force-unlock is performed. See [distributed lock diagnostics](distributed-e2e.md) for the limits of guard-drop and physical-release observations.
+
+Use `--diagnostics` only as a separately identified diagnostic run; trace overhead is not performance evidence. Keep initial failure, later convergence, and restarted-node checks separate. A retained run stops its owned nodes after capturing final evidence; its volumes and private files remain available for investigation. Successful runs can remove resources with `--cleanup-on-success`.
+
+```bash
+python3 scripts/repro_rejoin_readiness_docker.py cleanup \
+  --out-dir /path/to/private-evidence
+```
+
+Every data/configuration volume is verified before staging mounts or changes it; mounts disable image-content copying. Cleanup verifies every extant resource against the manifest's project, source and ownership labels before deleting any resource. Capture interruptions remain failures and continue through shutdown and summary writing; interrupted retention is explicitly reported as unconfirmed. It refuses foreign resources and does not prune the daemon. Output directories contain generated credentials, curl configurations and private CA/node keys: do not commit or upload them wholesale. Publish only a reviewed, redacted result summary with source/build identity and explicit environment limits.

scripts/README.md

@@ -47,6 +47,8 @@ their issue closes.
 
 | Entry | Status | Purpose | Wiring / docs |
 |---|---|---|---|
+| `repro_rejoin_readiness_docker.py` | dev-tool | Fixed-source Docker HTTPS rejoin and acknowledged-overwrite diagnostic; retains first failure | [Guide](../docs/testing/docker-rejoin-readiness.md) |
+| `test_repro_rejoin_readiness_docker.py` | ci-gate | Provenance, native architecture, ownership and failure-preservation guards | `make script-tests`; same guide |
 | `diagnose_scanner_enumeration_restart.py` | dev-tool | Strict fixed raw-entry-budget scanner-worker restart diagnostic | [Checkpoint fixture](../docs/testing/scanner-checkpoint-fixture.md) |
 | `prepare_replacement_migration.py` | dev-tool | Prepares digest-bound schema 5/6 replacement maintenance approvals | [Replacement recovery](../docs/operations/replacement-generation-recovery.md) |
 | `test_prepare_replacement_migration.py` | dev-tool | Verifies maintenance approval scope, publication, and stopped-writer assertion | Python unittest; same runbook |

scripts/repro_rejoin_readiness_docker.py

@@ -0,0 +1,875 @@
+#!/usr/bin/env python3
+# Copyright 2026 RustFS Team
+# Licensed under the Apache License, Version 2.0.
+"""Reproduce recovered-node admission on an isolated five-node HTTPS Docker pool.
+
+Run on a Linux AMD64 or ARM64 Docker daemon, with an image matching the daemon::
+
+    python3 scripts/repro_rejoin_readiness_docker.py run \
+        --image IMAGE --expected-source COMMIT_SHA --out-dir /tmp/rejoin-case \
+        --gate cluster --payload-mode distinct
+
+The image must carry io.rustfs.repro.source_revision and
+io.rustfs.repro.binary_sha256 labels. It is resolved to a local immutable image
+ID; nothing is pulled. The image needs Python 3, getent and /usr/bin/rustfs.
+The host needs Docker, OpenSSL and curl with built-in AWS SigV4 support.
+
+Evidence and generated credentials/CA keys live in a new private directory.
+That directory also retains the driver source snapshot and its SHA256. A failed
+immediate recovery batch is followed by one separate node-write-gated read-only
+convergence sample; that later sample never changes the original failure.
+Do not upload that directory wholesale. Failures retain stopped Docker resources
+after recovery and evidence capture, even with --cleanup-on-success. Explicit cleanup verifies every resource's exact
+name and ownership before deleting any::
+
+    python3 scripts/repro_rejoin_readiness_docker.py cleanup --out-dir /tmp/rejoin-case
+
+The evidence records the host and Docker-daemon architectures. All five nodes
+share one Docker daemon; their volumes do not establish independent physical
+failure domains. Image/daemon architecture mismatches are rejected rather than
+silently running the server under architecture emulation.
+"""
+
+from __future__ import annotations
+
+import argparse
+from concurrent.futures import ThreadPoolExecutor
+import hashlib
+import inspect
+import io
+import json
+import os
+from pathlib import Path
+import platform
+import re
+import secrets
+import subprocess
+import sys
+import tarfile
+import threading
+import time
+from datetime import datetime, timezone
+from typing import Any
+
+
+PROJECT_RE = re.compile(r"rustfs-rejoin-repro-[0-9a-f]{16}\Z")
+SOURCE_RE = re.compile(r"[0-9a-f]{40}\Z")
+IMAGE_RE = re.compile(r"sha256:[0-9a-f]{64}\Z")
+SHA_RE = re.compile(r"[0-9a-f]{64}\Z")
+LABEL_PREFIX = "io.rustfs.repro."
+OWNER = "rejoin-readiness-v1"
+PORT = 19000
+ORIGINAL_PAYLOAD = b"same payload for reference and each degraded overwrite\n"
+GATE_PATHS = {
+    "cluster": ("/minio/health/cluster",),
+    "node-write": ("/health/ready", "/minio/health/cluster"),
+}
+
+
+class ReproError(RuntimeError):
+    """A failed precondition or experiment assertion."""
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def write_json(path: Path, value: Any) -> None:
+    path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")
+    path.chmod(0o600)
+
+
+def sha256(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def elf_architecture(header: bytes, expected_arch: str = "") -> str:
+    """Validate a Linux executable's ELF64 header and optional native architecture."""
+    if len(header) < 64 or header[:7] != b"\x7fELF\x02\x01\x01":
+        raise ValueError("binary must have an ELF64 little-endian version-one header")
+    if (int.from_bytes(header[16:18], "little") not in (2, 3)
+            or int.from_bytes(header[20:24], "little") != 1
+            or int.from_bytes(header[52:54], "little") != 64):
+        raise ValueError("invalid executable ELF64 header")
+    machines = {183: "arm64", 62: "amd64"}
+    machine = int.from_bytes(header[18:20], "little")
+    if machine not in machines:
+        raise ValueError("unsupported ELF machine; expected AArch64 or x86_64")
+    architecture = machines[machine]
+    if expected_arch and architecture != expected_arch:
+        raise ValueError("binary ELF architecture does not match the Docker daemon architecture")
+    return architecture
+
+
+def payload_for(mode: str, generation: int, reference: bool = False) -> bytes:
+    if mode == "original":
+        return ORIGINAL_PAYLOAD
+    if mode != "distinct" or generation < 0:
+        raise ReproError("invalid payload mode or generation")
+    label = f"{'reference' if reference else 'overwrite'}:{generation:08d}:"
+    if len(label) >= len(ORIGINAL_PAYLOAD):
+        raise ReproError("payload generation is too large")
+    return label.ljust(len(ORIGINAL_PAYLOAD) - 1, ".").encode() + b"\n"
+
+
+def curl_quote(value: str) -> str:
+    if any(c in value for c in "\r\n\0"):
+        raise ReproError("curl configuration values cannot contain line breaks")
+    return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
+
+
+def parse_http_status(stdout: bytes) -> int:
+    value = stdout.decode("ascii", errors="replace").strip()
+    if not re.fullmatch(r"[0-9]{3}", value):
+        raise ReproError("curl returned an invalid HTTP status")
+    return int(value)
+
+
+def response_cause(status: int, headers: str, body: bytes) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    for line in headers.splitlines():
+        name, separator, value = line.partition(":")
+        if separator and name.lower() in ("retry-after", "x-rustfs-readiness-pending", "x-rustfs-error-cause", "x-amz-request-id"):
+            result[name.lower()] = value.strip()
+    text = body.decode("utf-8", errors="replace")
+    code = re.search(r"<Code>([^<]{1,120})</Code>", text)
+    if code:
+        result["s3_code"] = code.group(1)
+    if status == 503 and "x-rustfs-readiness-pending" in result:
+        result["classification"] = "startup_admission"
+    elif status == 503:
+        result["classification"] = "service_unavailable_without_startup_header"
+    elif status == 0:
+        result["classification"] = "transport_failure"
+    return result
+
+
+def canonical_architecture(value: Any) -> str:
+    aliases = {"aarch64": "arm64", "arm64": "arm64", "x86_64": "amd64", "amd64": "amd64"}
+    if not isinstance(value, str) or value not in aliases:
+        raise ReproError("unsupported architecture; the experiment requires AMD64 or ARM64")
+    return aliases[value]
+
+
+def validate_image(info: dict[str, Any], source: str, expected_binary: str | None = None,
+                   *, expected_arch: str) -> dict[str, str]:
+    if not SOURCE_RE.fullmatch(source):
+        raise ReproError("expected source must be a lowercase 40-character commit SHA")
+    image_id = info.get("Id", "")
+    labels = (info.get("Config") or {}).get("Labels") or {}
+    binary = labels.get(LABEL_PREFIX + "binary_sha256", "")
+    if not IMAGE_RE.fullmatch(image_id):
+        raise ReproError("image inspection did not return an immutable image ID")
+    architecture = canonical_architecture(info.get("Architecture"))
+    if info.get("Os") != "linux" or architecture != canonical_architecture(expected_arch):
+        raise ReproError("the experiment requires a Linux image matching the Docker daemon architecture")
+    inherited = (info.get("Config") or {}).get("Env") or ()
+    if any(entry.partition("=")[0].startswith("RUSTFS_") or entry.partition("=")[0] == "RUST_LOG"
+           for entry in inherited):
+        raise ReproError("the image must not inherit RUSTFS_* or RUST_LOG overrides")
+    if labels.get(LABEL_PREFIX + "source_revision") != source:
+        raise ReproError("image source revision does not match --expected-source")
+    if not SHA_RE.fullmatch(binary):
+        raise ReproError("image requires a valid binary SHA256 label")
+    if expected_binary is not None and binary != expected_binary:
+        raise ReproError("image binary SHA256 does not match the expected binary")
+    return {"id": image_id, "source_revision": source, "binary_sha256": binary, "architecture": architecture, "os": "linux",
+            "build_profile": labels.get(LABEL_PREFIX + "build_profile", "unknown")}
+
+
+def expected_resources(project: str) -> list[tuple[str, str, str]]:
+    if not PROJECT_RE.fullmatch(project):
+        raise ReproError("invalid experiment project name")
+    resources = [("container", f"{project}-init", "init")]
+    resources += [("container", f"{project}-storage{n:02}", "node") for n in range(1, 6)]
+    resources += [("volume", f"{project}-{kind}{n:02}", kind) for n in range(1, 6) for kind in ("data", "config")]
+    return resources + [("network", project, "network")]
+
+
+def resource_labels(project: str, source: str, kind: str) -> dict[str, str]:
+    return {LABEL_PREFIX + "owner": OWNER, LABEL_PREFIX + "project": project,
+            LABEL_PREFIX + "source_revision": source, LABEL_PREFIX + "kind": kind}
+
+
+def verify_ownership(info: dict[str, Any], resource: str, name: str, kind: str, project: str, source: str) -> None:
+    labels = ((info.get("Config") or {}).get("Labels") or {}) if resource == "container" else (info.get("Labels") or {})
+    actual_name = info.get("Name", "").removeprefix("/")
+    if actual_name != name or any(labels.get(key) != value for key, value in resource_labels(project, source, kind).items()):
+        raise ReproError(f"refusing to mutate unowned {resource} {name}")
+
+
+def is_missing_resource(resource: str, name: str, error: str) -> bool:
+    subject = re.escape(name.lower())
+    forms = {
+        "container": rf"no such container:\s*{subject}",
+        "volume": rf"(?:no such volume:\s*{subject}|get {subject}:\s*no such volume)",
+        "network": rf"(?:no such network:\s*{subject}|network {subject} not found)",
+        "image": rf"no such image:\s*{subject}(?::latest)?",
+    }
+    if resource not in forms:
+        return False
+    pattern = r"(?:error(?: response from daemon)?:\s*)?" + forms[resource]
+    lines = [line.strip().lower() for line in error.splitlines() if line.strip()]
+    return bool(lines) and all(re.fullmatch(pattern, line) is not None for line in lines)
+
+
+class Commands:
+    def __init__(self) -> None:
+        self.redactions: list[str] = []
+
+    def redact(self, value: str) -> str:
+        for secret in self.redactions:
+            value = value.replace(secret, "[REDACTED]")
+        return value
+
+    def run(self, args: list[str], *, data: bytes | None = None, timeout: float = 120,
+            check: bool = True) -> subprocess.CompletedProcess[bytes]:
+        try:
+            result = subprocess.run(args, input=data, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                                    timeout=timeout, check=False)
+        except (OSError, subprocess.TimeoutExpired) as error:
+            raise ReproError(f"{args[0]} {args[1] if len(args) > 1 else ''} failed: {type(error).__name__}") from error
+        if check and result.returncode:
+            detail = self.redact(result.stderr.decode(errors="replace"))[-1500:]
+            raise ReproError(f"{args[0]} {args[1] if len(args) > 1 else ''} failed ({result.returncode}): {detail}")
+        return result
+
+    def docker(self, *args: str, **kwargs: Any) -> subprocess.CompletedProcess[bytes]:
+        return self.run(["docker", *args], **kwargs)
+
+    def inspect(self, resource: str, name: str, *, optional: bool = False) -> dict[str, Any] | None:
+        result = self.docker(resource, "inspect", name, check=False)
+        if result.returncode:
+            error = result.stderr.decode(errors="replace").lower()
+            missing = is_missing_resource(resource, name, error)
+            if optional and missing:
+                return None
+            raise ReproError(f"could not inspect {resource} {name}")
+        value = json.loads(result.stdout)
+        if not isinstance(value, list) or len(value) != 1 or not isinstance(value[0], dict):
+            raise ReproError(f"unexpected inspection result for {resource}")
+        return value[0]
+
+
+def cleanup_resources(commands: Commands, manifest: dict[str, Any]) -> None:
+    project, source = manifest["project"], manifest["source_revision"]
+    if not isinstance(project, str) or not isinstance(source, str) or not SOURCE_RE.fullmatch(source):
+        raise ReproError("invalid source revision in manifest")
+    owned = []
+    # Verify the entire set before deleting anything; a bad manifest cannot
+    # partially tear down another experiment.
+    for resource, name, kind in expected_resources(project):
+        info = commands.inspect(resource, name, optional=True)
+        if info is not None:
+            verify_ownership(info, resource, name, kind, project, source)
+            owned.append((resource, name, info.get("Id", name)))
+    for resource, name, identifier in owned:
+        if resource == "container":
+            commands.docker("container", "rm", "--force", identifier)
+        elif resource == "network":
+            commands.docker(resource, "rm", identifier)
+        else:
+            commands.docker(resource, "rm", name)
+
+
+INIT_SCRIPT = r'''
+import hashlib, json, os, pathlib, stat, subprocess, sys, tarfile
+root = pathlib.Path('/stage')
+for n in range(1, 6):
+    for kind in ('data', 'config'):
+        path = root / f'{kind}{n:02}'
+        os.chmod(path, 0o700)
+        os.chown(path, 10001, 10001)
+with tarfile.open(fileobj=sys.stdin.buffer, mode='r|') as archive:
+    for member in archive:
+        parts = pathlib.PurePosixPath(member.name).parts
+        if (not member.isfile() or len(parts) != 3 or
+            parts[0] not in [f'config{n:02}' for n in range(1, 6)] or
+            parts[1] not in ('tls', 'secrets') or
+            parts[2] not in ('ca.crt', 'rustfs_cert.pem', 'rustfs_key.pem', 'access-key', 'secret-key')):
+            raise RuntimeError('unexpected configuration archive entry')
+        path = root.joinpath(*parts)
+        path.parent.mkdir(mode=0o700, exist_ok=True)
+        os.chown(path.parent, 10001, 10001)
+        data = archive.extractfile(member).read()
+        with open(path, 'xb') as output:
+            output.write(data)
+        os.chmod(path, 0o600)
+        os.chown(path, 10001, 10001)
+with open('/usr/bin/rustfs', 'rb') as binary:
+    binary_architecture = elf_architecture(binary.read(64), sys.argv[1])
+    binary.seek(0)
+    digest = hashlib.file_digest(binary, 'sha256').hexdigest() if hasattr(hashlib, 'file_digest') else None
+    if digest is None:
+        binary.seek(0)
+        h = hashlib.sha256()
+        for block in iter(lambda: binary.read(1024 * 1024), b''):
+            h.update(block)
+        digest = h.hexdigest()
+if digest != sys.argv[2]:
+    raise RuntimeError('binary SHA256 does not match its image label')
+filesystems = []
+for n in range(1, 6):
+    path = f'/stage/data{n:02}'
+    kind = subprocess.check_output(['stat', '-f', '-c', '%T', path], text=True).strip()
+    filesystems.append({'node': n, 'filesystem_type': kind, 'uid': os.stat(path).st_uid,
+                        'mode': oct(stat.S_IMODE(os.stat(path).st_mode))})
+os_release = {}
+with open('/etc/os-release') as release:
+    for line in release:
+        key, separator, value = line.strip().partition('=')
+        if separator and key in ('PRETTY_NAME', 'ID', 'VERSION_ID'):
+            os_release[key] = value.strip('"')
+version = subprocess.check_output(['/usr/bin/rustfs', '--version'], text=True).strip()
+print(json.dumps({'binary_sha256': digest, 'binary_architecture': binary_architecture, 'binary_version': version,
+                  'runtime_os': os_release, 'data_volumes': filesystems}))
+'''
+
+NODE_ENTRYPOINT = r'''
+set -eu
+attempt=0
+while :; do
+    all_ready=true
+    for peer in storage01 storage02 storage03 storage04 storage05; do
+        getent hosts "$peer" >/dev/null || all_ready=false
+    done
+    [ "$all_ready" = true ] && break
+    attempt=$((attempt + 1))
+    [ "$attempt" -lt 60 ] || exit 1
+    sleep 1
+done
+exec /usr/bin/rustfs
+'''
+
+
+class Experiment:
+    def __init__(self, args: argparse.Namespace, commands: Commands) -> None:
+        self.args, self.commands = args, commands
+        self.out = args.out_dir.resolve()
+        self.project = "rustfs-rejoin-repro-" + secrets.token_hex(8)
+        self.source = args.expected_source
+        self.nodes: dict[int, dict[str, Any]] = {}
+        self.acknowledged: dict[str, bytes] = {}
+        self.first_failure: dict[str, Any] | None = None
+        self.request_number = 0
+        self.record_lock = threading.Lock()
+        self.started = False
+        self.summary: dict[str, Any] = {"project": self.project, "source_revision": self.source,
+            "gate": args.gate, "payload_mode": args.payload_mode, "stress_iterations": args.stress_iterations,
+            "outage_wait_seconds": args.outage_wait, "host_platform": platform.system(), "host_architecture": platform.machine(),
+            "declared_build_profile": args.build_profile, "diagnostics_enabled": args.diagnostics,
+            "experiment_scope": "five HTTPS containers on one Docker daemon; their volumes do not establish independent physical failure domains",
+            "started_at": utc_now(), "object_request_retries": 0}
+
+    def event(self, value: dict[str, Any]) -> None:
+        with self.record_lock:
+            with (self.out / "events.jsonl").open("a") as output:
+                output.write(self.commands.redact(json.dumps(value, sort_keys=True)) + "\n")
+
+    def labels(self, kind: str) -> list[str]:
+        return [item for key, value in resource_labels(self.project, self.source, kind).items()
+                for item in ("--label", f"{key}={value}")]
+
+    def fail(self, phase: str, reason: str, request: dict[str, Any] | None = None) -> None:
+        failure = {"phase": phase, "reason": self.commands.redact(reason), "at": utc_now()}
+        if request:
+            failure["request"] = request
+        if self.first_failure is None:
+            self.first_failure = failure
+        self.event({"failure": failure})
+
+    def preflight(self) -> None:
+        self.out.mkdir(mode=0o700, parents=False, exist_ok=False)
+        os.umask(0o077)
+        private = self.out / "private"
+        private.mkdir(mode=0o700)
+        driver_source = Path(__file__).resolve().read_bytes()
+        driver_snapshot = private / "driver.py"
+        driver_snapshot.write_bytes(driver_source)
+        driver_snapshot.chmod(0o600)
+        self.summary["driver"] = {"source_sha256": sha256(driver_source), "snapshot": "private/driver.py"}
+        help_result = self.commands.run(["curl", "--disable", "--help", "all"])
+        if b"--aws-sigv4" not in help_result.stdout:
+            raise ReproError("host curl requires built-in --aws-sigv4 support")
+        self.summary["host_tools"] = {
+            "curl_version": self.commands.run(["curl", "--disable", "--version"]).stdout.decode(errors="replace").strip(),
+            "openssl_version": self.commands.run(["openssl", "version"]).stdout.decode(errors="replace").strip(),
+        }
+        server = json.loads(self.commands.docker("info", "--format", "{{json .}}").stdout)
+        if server.get("OSType") != "linux":
+            raise ReproError("this harness requires a Linux Docker daemon")
+        daemon_arch = canonical_architecture(server.get("Architecture"))
+        image = self.commands.inspect("image", self.args.image)
+        assert image is not None
+        self.identity = validate_image(image, self.source, self.args.expected_binary_sha256, expected_arch=daemon_arch)
+        if self.args.build_profile is not None and self.identity["build_profile"] != self.args.build_profile:
+            raise ReproError("image build profile label does not match --build-profile")
+        self.summary["image"] = self.identity
+        self.summary["docker"] = {key: server.get(key) for key in
+            ("ServerVersion", "OperatingSystem", "OSType", "Architecture", "KernelVersion", "Driver", "NCPU", "MemTotal")}
+        self.summary["docker"]["canonical_architecture"] = daemon_arch
+        self.summary["experiment_scope"] = (
+            f"five HTTPS {daemon_arch} containers on one Linux Docker daemon, controlled from "
+            f"{self.summary['host_platform']}/{self.summary['host_architecture']}; "
+            "their volumes do not establish independent physical failure domains"
+        )
+        self.summary["runtime_worker_threads"] = self.args.workers
+        write_json(self.out / "manifest.json", {"schema": 1, "project": self.project,
+            "source_revision": self.source, "image": self.identity})
+
+    def make_configuration(self) -> bytes:
+        private = self.out / "private"
+        self.access, self.secret = secrets.token_hex(12), secrets.token_hex(24)
+        self.commands.redactions = [self.secret, self.access]
+        (private / "access-key").write_text(self.access + "\n")
+        (private / "secret-key").write_text(self.secret + "\n")
+        self.commands.run(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "2", "-sha256",
+            "-subj", "/CN=RustFS-isolated-rejoin-CA", "-keyout", str(private / "ca.key"), "-out", str(private / "ca.crt")])
+        buffer = io.BytesIO()
+        with tarfile.open(fileobj=buffer, mode="w") as archive:
+            for n in range(1, 6):
+                node = f"storage{n:02}"
+                directory = private / node
+                directory.mkdir(mode=0o700)
+                self.commands.run(["openssl", "req", "-new", "-newkey", "rsa:2048", "-nodes", "-subj", f"/CN={node}",
+                    "-keyout", str(directory / "rustfs_key.pem"), "-out", str(directory / "request.csr")])
+                extensions = directory / "extensions"
+                extensions.write_text(f"subjectAltName=DNS:{node},IP:127.0.0.1\nextendedKeyUsage=serverAuth\n")
+                self.commands.run(["openssl", "x509", "-req", "-days", "2", "-sha256", "-in", str(directory / "request.csr"),
+                    "-CA", str(private / "ca.crt"), "-CAkey", str(private / "ca.key"), "-CAcreateserial", "-extfile", str(extensions),
+                    "-out", str(directory / "rustfs_cert.pem")])
+                for name, source in (("tls/ca.crt", private / "ca.crt"), ("tls/rustfs_cert.pem", directory / "rustfs_cert.pem"),
+                    ("tls/rustfs_key.pem", directory / "rustfs_key.pem"), ("secrets/access-key", private / "access-key"),
+                    ("secrets/secret-key", private / "secret-key")):
+                    data = source.read_bytes()
+                    info = tarfile.TarInfo(f"config{n:02}/{name}")
+                    info.size, info.mode, info.uid, info.gid = len(data), 0o600, 10001, 10001
+                    archive.addfile(info, io.BytesIO(data))
+        return buffer.getvalue()
+
+    def setup(self) -> None:
+        configuration = self.make_configuration()
+        self.commands.docker("network", "create", "--driver", "bridge", *self.labels("network"), self.project)
+        network = self.commands.inspect("network", self.project)
+        if network is None:
+            raise ReproError("could not inspect the experiment network")
+        verify_ownership(network, "network", self.project, "network", self.project, self.source)
+        self.summary["network"] = {key: network.get(key) for key in ("Id", "Name", "Driver", "Internal")}
+        if network.get("Driver") != "bridge" or network.get("Internal") is not False:
+            raise ReproError("the experiment requires its declared non-internal bridge network")
+        for n in range(1, 6):
+            for kind in ("data", "config"):
+                name = f"{self.project}-{kind}{n:02}"
+                self.commands.docker("volume", "create", *self.labels(kind), name)
+                volume = self.commands.inspect("volume", name)
+                if volume is None:
+                    raise ReproError("could not inspect a created experiment volume")
+                # Docker may reuse an existing name without applying labels.
+                # Validate every volume before staging can mount or chown it.
+                verify_ownership(volume, "volume", name, kind, self.project, self.source)
+        mounts = [item for n in range(1, 6) for kind in ("data", "config")
+                  for item in ("--mount", f"type=volume,src={self.project}-{kind}{n:02},dst=/stage/{kind}{n:02},volume-nocopy")]
+        init_name = self.project + "-init"
+        init_id = self.commands.docker("container", "create", "--pull=never", "--name", init_name, "--network", "none",
+            "--user", "0:0", "--interactive", *self.labels("init"), *mounts,
+            "--entrypoint", "python3", self.identity["id"], "-c", inspect.getsource(elf_architecture) + "\n" + INIT_SCRIPT,
+            self.identity["architecture"], self.identity["binary_sha256"]).stdout.decode().strip()
+        staged = json.loads(self.commands.docker("container", "start", "--attach", "--interactive", init_id, data=configuration).stdout)
+        if staged["binary_architecture"] != self.identity["architecture"]:
+            raise ReproError("the binary ELF architecture does not match the image and daemon")
+        if staged["binary_sha256"] != self.identity["binary_sha256"]:
+            raise ReproError("the image binary does not match its SHA256 label")
+        self.summary["staging"] = staged
+        info = self.commands.inspect("container", init_id)
+        assert info is not None
+        verify_ownership(info, "container", init_name, "init", self.project, self.source)
+        self.commands.docker("container", "rm", init_id)
+        environment = {
+            "RUSTFS_VOLUMES": " ".join(f"https://storage{n:02}:{PORT}/data" for n in range(1, 6)),
+            "RUSTFS_ADDRESS": f"0.0.0.0:{PORT}", "RUSTFS_CONSOLE_ADDRESS": "0.0.0.0:19001", "RUSTFS_CONSOLE_ENABLE": "true",
+            "RUSTFS_TLS_ENABLED": "true", "RUSTFS_TLS_PATH": "/config/tls", "RUSTFS_EXTRA_CA_CERT": "/config/tls/ca.crt",
+            "RUSTFS_STORAGE_CLASS_STANDARD": "EC:2", "RUSTFS_STORAGE_CLASS_RRS": "EC:2", "RUSTFS_ERASURE_SET_DRIVE_COUNT": "5",
+            "RUSTFS_ACCESS_KEY_FILE": "/config/secrets/access-key", "RUSTFS_SECRET_KEY_FILE": "/config/secrets/secret-key",
+            "RUSTFS_OBS_LOG_DIRECTORY": "/logs", "RUSTFS_OBS_LOGGER_LEVEL": "warn", "RUSTFS_HEAL_MAX_CONCURRENT_HEALS": "1",
+            "RUSTFS_HEAL_MAX_CONCURRENT_PER_SET": "1", "RUSTFS_SCANNER_SPEED": "slow", "RUSTFS_SCANNER_IDLE_MODE": "true",
+            "RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS": "1", "RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS": "1",
+            "RUSTFS_HEALTH_CLUSTER_TIMEOUT_MS": "15000",
+        }
+        if self.args.workers is not None:
+            environment["RUSTFS_RUNTIME_WORKER_THREADS"] = str(self.args.workers)
+        if self.args.diagnostics:
+            environment.update({"RUSTFS_OBJECT_LOCK_DIAG_ENABLE": "true", "RUSTFS_OBJECT_LOCK_DIAG_SLOW_ACQUIRE_MS": "100",
+                "RUSTFS_OBJECT_LOCK_DIAG_SLOW_HOLD_MS": "100",
+                "RUST_LOG": "warn,rustfs_ecstore::object_lock_diag=trace,rustfs_lock::namespace_owner=trace,"
+                            "rustfs_ecstore::cluster::rpc::remote_locker=debug,rustfs_lock::distributed_lock=debug"})
+        self.summary["environment"] = environment
+        for n in range(1, 6):
+            name = f"{self.project}-storage{n:02}"
+            args = ["container", "create", "--pull=never", "--name", name, "--hostname", f"storage{n:02}",
+                *self.labels("node"), "--network", self.project, "--network-alias", f"storage{n:02}", "--user", "10001:10001",
+                "--read-only", "--cpus", "1", "--memory", "1g", "--cap-drop", "ALL", "--security-opt", "no-new-privileges:true",
+                "--tmpfs", "/tmp:mode=1777,size=64m", "--tmpfs", "/logs:uid=10001,gid=10001,mode=0700,size=32m",
+                "--publish", f"127.0.0.1::{PORT}", "--mount", f"type=volume,src={self.project}-data{n:02},dst=/data,volume-nocopy",
+                "--mount", f"type=volume,src={self.project}-config{n:02},dst=/config,readonly,volume-nocopy"]
+            args += [item for key, value in environment.items() for item in ("--env", f"{key}={value}")]
+            args += ["--entrypoint", "/bin/sh", self.identity["id"], "-c", NODE_ENTRYPOINT]
+            identifier = self.commands.docker(*args).stdout.decode().strip()
+            if not re.fullmatch(r"[0-9a-f]{64}", identifier):
+                raise ReproError("Docker returned an invalid container ID")
+            self.nodes[n] = {"id": identifier, "name": name}
+        self.started = True
+        self.commands.docker("container", "start", *(node["id"] for node in self.nodes.values()))
+        self.snapshot("initial-before-gate")
+
+    def snapshot(self, phase: str) -> None:
+        snapshot = []
+        for n, node in self.nodes.items():
+            info = self.commands.inspect("container", node["id"])
+            assert info is not None
+            verify_ownership(info, "container", node["name"], "node", self.project, self.source)
+            bindings = (info.get("NetworkSettings", {}).get("Ports", {}).get(f"{PORT}/tcp") or [])
+            node["running"] = info["State"]["Running"]
+            if node["running"]:
+                if len(bindings) != 1 or bindings[0].get("HostIp") != "127.0.0.1":
+                    raise ReproError("S3 must be published on one loopback-only ephemeral port")
+                node["port"] = int(bindings[0]["HostPort"])
+            node["generation"] = {"container_id": info["Id"], "started_at": info["State"]["StartedAt"],
+                                  "restart_count": info.get("RestartCount")}
+            limits = info["HostConfig"]
+            snapshot.append({"node": n, "image": info["Image"], "generation": node["generation"], "state": info["State"],
+                "limits": {key: limits.get(key) for key in ("NanoCpus", "Memory", "ReadonlyRootfs", "CapDrop", "SecurityOpt", "Tmpfs")},
+                "mounts": [{key: mount.get(key) for key in ("Type", "Name", "Destination", "RW", "Driver")} for mount in info["Mounts"]]})
+        write_json(self.out / f"containers-{phase}.json", snapshot)
+
+    def request(self, node: int, method: str, path: str, *, payload: bytes | None = None,
+                signed: bool = True, phase: str = "probe", timeout: float = 30) -> dict[str, Any]:
+        with self.record_lock:
+            self.request_number += 1
+            request_id = self.request_number
+        prefix = f"request-{request_id:05}"
+        port = self.nodes[node]["port"]
+        url = f"https://storage{node:02}:{port}{path}"
+        resolution = f"storage{node:02}:{port}:127.0.0.1"
+        body_file, header_file = self.out / f"{prefix}.body", self.out / f"{prefix}.headers"
+        config = ["silent", "show-error", 'proto = "=https"', 'noproxy = "*"', "connect-timeout = 5", f"max-time = {timeout}",
+            "retry = 0", f"url = {curl_quote(url)}", f"resolve = {curl_quote(resolution)}",
+            f"cacert = {curl_quote(str(self.out / 'private/ca.crt'))}", f"request = {curl_quote(method)}",
+            f"output = {curl_quote(str(body_file))}", f"dump-header = {curl_quote(str(header_file))}", 'write-out = "%{http_code}"']
+        if signed:
+            config += ['aws-sigv4 = "aws:amz:us-east-1:s3"', f"user = {curl_quote(self.access + ':' + self.secret)}",
+                       f"header = {curl_quote('x-amz-content-sha256: ' + sha256(payload or b''))}"]
+        if payload is not None:
+            payload_file = self.out / "private" / f"{prefix}.payload"
+            payload_file.write_bytes(payload)
+            config.append(f"upload-file = {curl_quote(str(payload_file))}")
+        config_file = self.out / "private" / f"{prefix}.curl"
+        config_file.write_text("\n".join(config) + "\n")
+        began, wall = time.monotonic(), utc_now()
+        result = self.commands.run(["curl", "--disable", "--config", str(config_file)], check=False, timeout=timeout + 10)
+        status = parse_http_status(result.stdout)
+        body = body_file.read_bytes() if body_file.exists() else b""
+        headers = header_file.read_text(errors="replace") if header_file.exists() else ""
+        for secret in self.commands.redactions:
+            body = body.replace(secret.encode(), b"[REDACTED]")
+        headers = self.commands.redact(headers)
+        body_file.write_bytes(body)
+        header_file.write_text(headers)
+        record = {"id": request_id, "phase": phase, "node": node, "method": method, "path": path,
+            "http": status, "curl_exit": result.returncode, "started_at": wall, "duration_seconds": time.monotonic() - began,
+            "body": body_file.name, "headers": header_file.name, "body_sha256": sha256(body),
+            "generation": self.nodes[node]["generation"], "cause": response_cause(status, headers, body)}
+        if result.stderr:
+            record["transport_error"] = self.commands.redact(result.stderr.decode(errors="replace"))[-1000:]
+        self.event(record)
+        return record
+
+    def wait_gate(self, phase: str, *, gate: str | None = None) -> dict[str, Any]:
+        selected_gate = self.args.gate if gate is None else gate
+        deadline = time.monotonic() + self.args.gate_timeout
+        while time.monotonic() < deadline:
+            all_ready = True
+            for node in range(1, 6):
+                for path in GATE_PATHS[selected_gate]:
+                    remaining = deadline - time.monotonic()
+                    if remaining <= 0:
+                        raise ReproError(f"{phase} gate timed out")
+                    result = self.request(node, "GET", path, signed=False, phase=phase + "-gate", timeout=min(20, remaining))
+                    all_ready &= result["curl_exit"] == 0 and result["http"] == 200
+            if all_ready:
+                # No sleep, diagnostic probe, or Docker inspection may be added
+                # between this return and the immediate recovery object reads.
+                return {"gate": selected_gate, "paths": list(GATE_PATHS[selected_gate]), "completed_at": utc_now(),
+                        "monotonic": time.monotonic(), "last_request_id": self.request_number}
+            time.sleep(min(2, max(0, deadline - time.monotonic())))
+        raise ReproError(f"{phase} gate timed out")
+
+    def require(self, record: dict[str, Any], phase: str, expected: bytes | None = None) -> None:
+        reason = None
+        if record["curl_exit"] or record["http"] != 200:
+            reason = "object request failed"
+        elif expected is not None and record["body_sha256"] != sha256(expected):
+            reason = "object payload did not match the last acknowledged payload"
+        if reason:
+            self.fail(phase, reason, record)
+            raise ReproError(reason)
+
+    def sequence(self) -> None:
+        self.summary["initial_gate"] = self.wait_gate("initial")
+        deadline = time.monotonic() + self.args.gate_timeout
+        while True:
+            result = self.request(1, "PUT", "/repro", phase="create-bucket")
+            if result["http"] == 200 and not result["curl_exit"]:
+                break
+            if result["http"] != 503 or result["curl_exit"] or time.monotonic() >= deadline:
+                self.require(result, "create-bucket")
+            time.sleep(2)
+        reference = payload_for(self.args.payload_mode, 0, reference=True)
+        self.require(self.request(1, "PUT", "/repro/reference-object", payload=reference, phase="initial-put"), "initial-put")
+        self.acknowledged["reference-object"] = reference
+        self.require(self.request(3, "GET", "/repro/reference-object", phase="initial-get"), "initial-get", reference)
+        generation = 0
+        for stopped, peer in ((5, 2), (2, 3)):
+            self.commands.docker("container", "stop", "-t", "10", self.nodes[stopped]["id"], timeout=30)
+            time.sleep(self.args.outage_wait)
+            self.require(self.request(1, "GET", "/repro/reference-object", phase=f"down-{stopped}-reference"), "degraded-reference", reference)
+            for iteration in range(self.args.stress_iterations):
+                generation += 1
+                payload = payload_for(self.args.payload_mode, generation)
+                phase = f"down-{stopped}-iteration-{iteration + 1}"
+                self.require(self.request(1, "PUT", "/repro/degraded-object", payload=payload, phase=phase), phase)
+                self.acknowledged["degraded-object"] = payload
+                self.require(self.request(peer, "GET", "/repro/degraded-object", phase=phase), phase, payload)
+        self.summary["sequence_passed"] = True
+
+    def read_object_batch(self, phase: str, gate: dict[str, Any]) -> dict[str, Any]:
+        started, wall = time.monotonic(), utc_now()
+        failed = []
+        for key, expected in self.acknowledged.items():
+            for node in range(1, 6):
+                record = self.request(node, "GET", f"/repro/{key}", phase=phase)
+                try:
+                    self.require(record, phase, expected)
+                except ReproError:
+                    failed.append(record["id"])
+        return {"failed_request_ids": failed, "count": len(self.acknowledged) * 5,
+                "result": "failed" if failed else "passed", "started_at": wall, "finished_at": utc_now(),
+                "duration_seconds": time.monotonic() - started,
+                "gate_to_batch_start_seconds": started - gate["monotonic"]}
+
+    def recover(self) -> None:
+        if not self.started:
+            self.summary["recovery"] = "not started: setup did not complete"
+            return
+        self.snapshot("before-recovery")
+        if self.first_failure:
+            # Observe the still-failing generation before any stopped peer is
+            # restarted or log copying delays volatile lock observations.
+            failure_request = self.first_failure.get("request", {})
+            cause = failure_request.get("cause", {})
+            if cause.get("retry-after") == "1" or cause.get("x-rustfs-error-cause") in (
+                    "namespace_lock_timeout", "namespace_lock_unavailable"):
+                for sample in range(self.args.lock_samples):
+                    with ThreadPoolExecutor(max_workers=5) as executor:
+                        futures = [executor.submit(self.request, node, "GET", "/rustfs/admin/v3/top/locks?count=1000",
+                            phase=f"pre-recovery-locks-{sample + 1}", timeout=10)
+                            for node in self.nodes if self.nodes[node]["running"]]
+                        for future in futures:
+                            try:
+                                record = future.result()
+                                status = record["http"]
+                                self.event({"lock_snapshot_request_id": record["id"], "sample": sample + 1,
+                                    "api_result": "unsupported_or_not_registered" if status in (404, 405, 501)
+                                    else "available" if status == 200 else "request_failed", "http": status})
+                            except ReproError as error:
+                                self.event({"lock_snapshot_error": str(error), "sample": sample + 1})
+                    if sample + 1 < self.args.lock_samples:
+                        time.sleep(1)
+            try:
+                self.capture("before-recovery")
+            except (Exception, KeyboardInterrupt) as error:
+                self.fail("pre-recovery-capture", str(error) or "interrupted")
+        restart = [node["id"] for node in self.nodes.values() if not node["running"]]
+        if restart:
+            self.commands.docker("container", "start", *restart)
+        self.snapshot("restarted-before-gate")
+        self.summary["recovery_gate"] = self.wait_gate("recovery")
+        self.summary["recovery_gets"] = self.read_object_batch("immediate-recovery", self.summary["recovery_gate"])
+        # Classification probes run only after the immediate object batch.
+        for node in range(1, 6):
+            for path in ("/health/ready", "/minio/health/cluster", "/minio/health/cluster/read"):
+                self.request(node, "GET", path, signed=False, phase="post-recovery-classification", timeout=20)
+        if self.summary["recovery_gets"]["failed_request_ids"]:
+            convergence = {"started_at": utc_now()}
+            self.summary["convergence"] = convergence
+            try:
+                convergence["gate"] = self.wait_gate("convergence", gate="node-write")
+            except ReproError as error:
+                self.fail("convergence-gate", str(error))
+                convergence.update(result="gate_failed", finished_at=utc_now())
+                return
+            convergence["gets"] = self.read_object_batch("later-convergence", convergence["gate"])
+            convergence.update(result=convergence["gets"]["result"], finished_at=utc_now())
+
+    def capture(self, phase: str = "final") -> None:
+        for node in self.nodes.values():
+            result = self.commands.docker("container", "logs", "--timestamps", "--tail", "2000", node["id"], check=False)
+            if result.returncode:
+                self.fail("capture", f"Docker stdout capture failed for {node['name']}: "
+                          + self.commands.redact(result.stderr.decode(errors="replace"))[-500:])
+            (self.out / (node["name"].rsplit("-", 1)[-1] + f"-{phase}.log")).write_text(
+                self.commands.redact((result.stdout + result.stderr).decode(errors="replace")))
+            archive = self.commands.docker("container", "cp", node["id"] + ":/logs/.", "-", check=False)
+            if archive.returncode:
+                self.fail("capture", f"runtime log capture unavailable for {node['name']} during {phase}: "
+                          + self.commands.redact(archive.stderr.decode(errors="replace"))[-500:])
+                continue
+            log_dir = self.out / f"runtime-logs-{phase}" / node["name"].rsplit("-", 1)[-1]
+            log_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
+            with tarfile.open(fileobj=io.BytesIO(archive.stdout)) as logs:
+                for member in logs:
+                    if member.isfile() and member.size <= 32 * 1024 * 1024:
+                        relative = Path(member.name)
+                        if relative.is_absolute() or ".." in relative.parts:
+                            raise ReproError("unexpected runtime log archive path")
+                        target = log_dir / relative
+                        target.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+                        data = logs.extractfile(member).read()
+                        for secret in self.commands.redactions:
+                            data = data.replace(secret.encode(), b"[REDACTED]")
+                        target.write_bytes(data)
+        if self.started:
+            self.snapshot("final")
+
+    def stop_retained_nodes(self, *, allow_missing: bool = False) -> None:
+        running = []
+        existing = []
+        by_name = {node["name"]: node for node in self.nodes.values()}
+        # Include the staging container and partially created nodes. A Docker
+        # CLI timeout can leave an owned container alive before its ID is saved.
+        # Verify the complete inventory before stopping any immutable ID.
+        for resource, name, kind in expected_resources(self.project):
+            if resource != "container":
+                continue
+            node = by_name.get(name)
+            optional = allow_missing or node is None
+            info = self.commands.inspect("container", node["id"] if node else name, optional=optional)
+            if info is None:
+                if optional:
+                    continue
+                raise ReproError("a retained experiment node could not be inspected")
+            verify_ownership(info, "container", name, kind, self.project, self.source)
+            existing.append((info["Id"], name, kind))
+            if info["State"]["Running"]:
+                running.append(info["Id"])
+        if running:
+            self.commands.docker("container", "stop", "-t", "10", *running, timeout=80)
+        snapshot = []
+        for identifier, name, kind in existing:
+            info = self.commands.inspect("container", identifier)
+            if info is None:
+                raise ReproError("could not confirm a retained container stopped")
+            verify_ownership(info, "container", name, kind, self.project, self.source)
+            snapshot.append({"id": identifier, "name": name, "kind": kind, "state": info["State"]})
+            if name in by_name:
+                by_name[name]["running"] = info["State"]["Running"]
+        write_json(self.out / "containers-retained-stopped.json", snapshot)
+        if any(info["state"]["Running"] for info in snapshot):
+            raise ReproError("an owned experiment container is still running after retention stop")
+        self.summary["retained_nodes_stopped"] = True
+
+    def run(self) -> int:
+        self.preflight()
+        try:
+            self.setup()
+            self.sequence()
+        except (Exception, KeyboardInterrupt) as error:
+            self.fail("sequence-or-setup", str(error) or "interrupted")
+        finally:
+            try:
+                self.recover()
+            except (Exception, KeyboardInterrupt) as error:
+                self.fail("recovery", str(error) or "interrupted")
+            try:
+                self.capture()
+            except (Exception, KeyboardInterrupt) as error:
+                self.fail("capture", str(error) or "interrupted")
+            self.summary["acknowledged_objects"] = {key: sha256(value) for key, value in self.acknowledged.items()}
+            self.summary["resources_retained"] = True
+            cleanup_failed = False
+            if self.first_failure is None and self.args.cleanup_on_success:
+                try:
+                    cleanup_resources(self.commands, json.loads((self.out / "manifest.json").read_text()))
+                    self.summary["resources_retained"] = False
+                except (Exception, KeyboardInterrupt) as error:
+                    cleanup_failed = True
+                    self.fail("cleanup", str(error) or "interrupted")
+                    self.summary["cleanup"] = "failed; resources may be partially removed"
+            if self.summary["resources_retained"]:
+                try:
+                    self.stop_retained_nodes(allow_missing=cleanup_failed)
+                except (Exception, KeyboardInterrupt) as error:
+                    self.fail("retention-stop", str(error) or "interrupted")
+                    self.summary["retained_nodes_stopped"] = False
+            self.summary["first_failure"] = self.first_failure
+            self.summary["result"] = "failed" if self.first_failure else "passed"
+            self.summary["finished_at"] = utc_now()
+            write_json(self.out / "summary.json", self.summary)
+        print(json.dumps({"result": self.summary["result"], "project": self.project,
+                          "resources_retained": self.summary["resources_retained"], "summary": str(self.out / "summary.json")}))
+        return 2 if self.first_failure else 0
+
+
+def positive_int(value: str) -> int:
+    result = int(value)
+    if result < 1:
+        raise argparse.ArgumentTypeError("value must be positive")
+    return result
+
+
+def lock_sample_count(value: str) -> int:
+    result = positive_int(value)
+    if result > 5:
+        raise argparse.ArgumentTypeError("lock snapshots are bounded to at most five samples")
+    return result
+
+
+def parser() -> argparse.ArgumentParser:
+    result = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    sub = result.add_subparsers(dest="command", required=True)
+    run = sub.add_parser("run", help="create an isolated pool and run the fault/recovery experiment")
+    run.add_argument("--image", required=True)
+    run.add_argument("--expected-source", required=True)
+    run.add_argument("--expected-binary-sha256")
+    run.add_argument("--build-profile", choices=("dev", "debug", "release"), help="require this image build-profile label; the actual label is always recorded")
+    run.add_argument("--diagnostics", action="store_true", help="enable the bounded lock diagnostic profile; default keeps warn logging")
+    run.add_argument("--lock-samples", type=lock_sample_count, default=2, help="bounded pre-recovery lock snapshots after Retry-After: 1")
+    run.add_argument("--out-dir", type=Path, required=True, help="new private evidence directory (must not exist)")
+    run.add_argument("--gate", choices=GATE_PATHS, default="cluster")
+    run.add_argument("--payload-mode", choices=("original", "distinct"), default="distinct")
+    run.add_argument("--stress-iterations", type=positive_int, default=1, help="overwrite/read pairs per outage")
+    run.add_argument("--workers", type=positive_int, help="set only for the two-worker comparison")
+    run.add_argument("--outage-wait", type=positive_int, default=40)
+    run.add_argument("--gate-timeout", type=positive_int, default=300)
+    run.add_argument("--cleanup-on-success", action="store_true")
+    clean = sub.add_parser("cleanup", help="remove only resources with this evidence manifest's exact ownership")
+    clean.add_argument("--out-dir", type=Path, required=True)
+    return result
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parser().parse_args(argv)
+    commands = Commands()
+    try:
+        if args.command == "cleanup":
+            manifest = json.loads((args.out_dir / "manifest.json").read_text())
+            cleanup_resources(commands, manifest)
+            print(json.dumps({"cleaned_project": manifest["project"]}))
+            return 0
+        return Experiment(args, commands).run()
+    except (ReproError, OSError, ValueError, KeyError) as error:
+        print(commands.redact(f"reproduction failed: {error}"), file=sys.stderr)
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/test_repro_rejoin_readiness_docker.py

@@ -0,0 +1,677 @@
+#!/usr/bin/env python3
+# Copyright 2026 RustFS Team
+# Licensed under the Apache License, Version 2.0.
+"""Guard and failure-path tests for the Docker rejoin reproduction harness."""
+
+import contextlib
+import io
+import inspect
+import json
+from pathlib import Path
+import re
+import subprocess
+import tempfile
+import time
+import unittest
+from unittest.mock import patch
+
+import repro_rejoin_readiness_docker as repro
+
+
+SOURCE = "a" * 40
+BINARY = "b" * 64
+IMAGE = "sha256:" + "c" * 64
+PROJECT = "rustfs-rejoin-repro-0123456789abcdef"
+
+
+def image_info():
+    return {"Id": IMAGE, "Os": "linux", "Architecture": "arm64", "Config": {"Labels": {
+        repro.LABEL_PREFIX + "source_revision": SOURCE, repro.LABEL_PREFIX + "binary_sha256": BINARY}}}
+
+
+class FakeDocker(repro.Commands):
+    def __init__(self, objects=None):
+        super().__init__()
+        self.objects = objects or {}
+        self.mutations = []
+
+    def inspect(self, resource, name, *, optional=False):
+        direct = self.objects.get((resource, name))
+        if direct is not None:
+            return direct
+        return next((value for (kind, _), value in self.objects.items() if kind == resource and value.get("Id") == name), None)
+
+    def docker(self, *args, **kwargs):
+        self.mutations.append(args)
+        if args[:2] == ("container", "stop"):
+            for info in self.objects.values():
+                if info.get("Id") in args[4:]:
+                    info["State"]["Running"] = False
+        return subprocess.CompletedProcess(args, 0, b"", b"")
+
+
+class FakeCurl(repro.Commands):
+    def __init__(self, payload=b"ok", status=200, headers=""):
+        super().__init__()
+        self.payload, self.status, self.headers = payload, status, headers
+        self.calls = []
+        self.configs = []
+
+    def run(self, args, **kwargs):
+        self.calls.append(args)
+        config = Path(args[-1]).read_text()
+        self.configs.append(config)
+        for key, content in (("output", self.payload), ("dump-header", self.headers.encode())):
+            value = re.search(r"^" + key + r" = (.+)$", config, re.MULTILINE).group(1)
+            Path(json.loads(value)).write_bytes(content)
+        return subprocess.CompletedProcess(args, 0, str(self.status).encode(), b"")
+
+
+def experiment(directory, commands):
+    args = repro.parser().parse_args(["run", "--image", IMAGE, "--expected-source", SOURCE,
+                                     "--out-dir", str(directory)])
+    value = repro.Experiment(args, commands)
+    value.out.mkdir(mode=0o700)
+    (value.out / "private").mkdir(mode=0o700)
+    value.access, value.secret = "test-access-private", "test-secret-private"
+    value.commands.redactions = [value.secret, value.access]
+    value.nodes = {n: {"id": str(n) * 64, "name": f"{value.project}-storage{n:02}", "port": 19000 + n,
+                       "generation": {"container_id": str(n) * 64, "started_at": "generation-one"}}
+                   for n in range(1, 6)}
+    return value
+
+
+class ImageGuards(unittest.TestCase):
+    def elf_header(self, machine):
+        header = bytearray(64)
+        header[:7] = b"\x7fELF\x02\x01\x01"
+        header[16:18] = (3).to_bytes(2, "little")
+        header[18:20] = machine.to_bytes(2, "little")
+        header[20:24] = (1).to_bytes(4, "little")
+        header[52:54] = (64).to_bytes(2, "little")
+        return bytes(header)
+
+    def test_binary_elf_header_requires_the_actual_native_machine(self):
+        for machine, architecture in ((183, "arm64"), (62, "amd64")):
+            with self.subTest(architecture=architecture):
+                header = self.elf_header(machine)
+                self.assertEqual(repro.elf_architecture(header), architecture)
+                self.assertEqual(repro.elf_architecture(header, architecture), architecture)
+                mismatch = "amd64" if architecture == "arm64" else "arm64"
+                with self.assertRaisesRegex(ValueError, "does not match"):
+                    repro.elf_architecture(header, mismatch)
+        for header in (b"not ELF", self.elf_header(183)[:20], b"MZ" + self.elf_header(183)[2:],
+                       self.elf_header(183)[:4] + b"\x01" + self.elf_header(183)[5:],
+                       self.elf_header(183)[:5] + b"\x02" + self.elf_header(183)[6:], self.elf_header(243)):
+            with self.subTest(header=header[:20]), self.assertRaises(ValueError):
+                repro.elf_architecture(header)
+
+    def test_same_elf_parser_can_run_in_the_isolated_init_script(self):
+        namespace = {}
+        exec(inspect.getsource(repro.elf_architecture), namespace)
+        self.assertEqual(namespace["elf_architecture"](self.elf_header(62), "amd64"), "amd64")
+        with self.assertRaisesRegex(ValueError, "does not match"):
+            namespace["elf_architecture"](self.elf_header(62), "arm64")
+
+    def test_resolves_only_the_expected_linux_source_and_binary(self):
+        self.assertEqual(repro.validate_image(image_info(), SOURCE, BINARY, expected_arch="arm64")["id"], IMAGE)
+        for change in ({"Os": "darwin"}, {"Id": "mutable-tag"}):
+            with self.subTest(change=change), self.assertRaises(repro.ReproError):
+                repro.validate_image({**image_info(), **change}, SOURCE, expected_arch="arm64")
+        with self.assertRaises(repro.ReproError):
+            repro.validate_image(image_info(), "d" * 40, expected_arch="arm64")
+        with self.assertRaises(repro.ReproError):
+            repro.validate_image(image_info(), SOURCE, "e" * 64, expected_arch="arm64")
+        missing_label = image_info()
+        del missing_label["Config"]["Labels"][repro.LABEL_PREFIX + "binary_sha256"]
+        with self.assertRaises(repro.ReproError):
+            repro.validate_image(missing_label, SOURCE, expected_arch="arm64")
+
+    def test_native_architecture_aliases_accept_matching_arm64_and_amd64(self):
+        for alias, expected in (("aarch64", "arm64"), ("arm64", "arm64"), ("x86_64", "amd64"), ("amd64", "amd64")):
+            with self.subTest(alias=alias):
+                self.assertEqual(repro.canonical_architecture(alias), expected)
+                identity = repro.validate_image({**image_info(), "Architecture": expected}, SOURCE, expected_arch=alias)
+                self.assertEqual(identity["architecture"], expected)
+
+    def test_mismatched_or_unsupported_architectures_are_rejected(self):
+        for image_arch, daemon_arch in (("arm64", "amd64"), ("amd64", "arm64"),
+                                        ("s390x", "s390x"), ("arm64", "ppc64le"), (None, "arm64")):
+            with self.subTest(image_arch=image_arch, daemon_arch=daemon_arch), self.assertRaises(repro.ReproError):
+                repro.validate_image({**image_info(), "Architecture": image_arch}, SOURCE, expected_arch=daemon_arch)
+
+    def test_inherited_controls_are_rejected_without_exposing_values(self):
+        for environment in (["RUSTFS_RUNTIME_WORKER_THREADS=2"], ["RUST_LOG=trace"], ["RUSTFS_SECRET_KEY=private-base-secret"]):
+            info = image_info()
+            info["Config"]["Env"] = environment
+            with self.subTest(environment=environment), self.assertRaises(repro.ReproError) as error:
+                repro.validate_image(info, SOURCE, expected_arch="arm64")
+            self.assertNotIn(environment[0], str(error.exception))
+            self.assertNotIn("private-base-secret", str(error.exception))
+        neutral = image_info()
+        neutral["Config"]["Env"] = ["PATH=/usr/bin", "LANG=C.UTF-8", "PYTHON_VERSION=3.13", "RUST_LOG_STYLE=never"]
+        self.assertEqual(repro.validate_image(neutral, SOURCE, expected_arch="arm64")["id"], IMAGE)
+
+    def test_preflight_uses_daemon_architecture_and_records_the_actual_host(self):
+        class PreflightCommands(repro.Commands):
+            def run(self, args, **kwargs):
+                output = {("curl", "--disable", "--help", "all"): b"--aws-sigv4",
+                          ("curl", "--disable", "--version"): b"curl TEST with TLS\n",
+                          ("openssl", "version"): b"OpenSSL TEST\n"}[tuple(args)]
+                return subprocess.CompletedProcess(args, 0, output, b"")
+
+            def docker(self, *args, **kwargs):
+                if args != ("info", "--format", "{{json .}}"):
+                    raise AssertionError("preflight must not mutate Docker resources")
+                return subprocess.CompletedProcess(args, 0, json.dumps({"OSType": "linux", "Architecture": "x86_64"}).encode(), b"")
+
+            def inspect(self, resource, name, **kwargs):
+                self.image_arch = getattr(self, "image_arch", "amd64")
+                return {**image_info(), "Architecture": self.image_arch}
+
+        with tempfile.TemporaryDirectory() as root:
+            args = repro.parser().parse_args(["run", "--image", IMAGE, "--expected-source", SOURCE,
+                                             "--out-dir", str(Path(root) / "matched")])
+            commands = PreflightCommands()
+            with patch.object(repro.platform, "system", return_value="Linux"), \
+                 patch.object(repro.platform, "machine", return_value="x86_64"), patch.object(repro.os, "umask"):
+                value = repro.Experiment(args, commands)
+                value.preflight()
+                self.assertEqual(value.summary["docker"]["canonical_architecture"], "amd64")
+                self.assertIn("HTTPS amd64", value.summary["experiment_scope"])
+                self.assertIn("Linux/x86_64", value.summary["experiment_scope"])
+                self.assertNotIn("not native AMD64", value.summary["experiment_scope"])
+                snapshot = (value.out / value.summary["driver"]["snapshot"]).read_bytes()
+                self.assertEqual(snapshot, Path(repro.__file__).read_bytes())
+                self.assertEqual(value.summary["driver"]["source_sha256"], repro.sha256(snapshot))
+                self.assertEqual((value.out / "private/driver.py").stat().st_mode & 0o777, 0o600)
+                self.assertEqual(value.summary["host_tools"], {"curl_version": "curl TEST with TLS", "openssl_version": "OpenSSL TEST"})
+                args.out_dir = Path(root) / "mismatched"
+                commands.image_arch = "arm64"
+                with self.assertRaises(repro.ReproError):
+                    repro.Experiment(args, commands).preflight()
+
+    def test_invalid_revision_and_project_never_authorize_cleanup(self):
+        docker = FakeDocker()
+        for project, source in (("other-project", SOURCE), (PROJECT + ";other", SOURCE), (PROJECT, "main"), (PROJECT, 123), (None, SOURCE)):
+            with self.subTest(project=project), self.assertRaises(repro.ReproError):
+                repro.cleanup_resources(docker, {"project": project, "source_revision": source})
+        self.assertEqual(docker.mutations, [])
+
+
+class CleanupGuards(unittest.TestCase):
+    def owned(self):
+        objects = {}
+        for resource, name, kind in repro.expected_resources(PROJECT):
+            info = {"Name": "/" + name if resource == "container" else name, "Id": name + "-immutable-id"}
+            labels = repro.resource_labels(PROJECT, SOURCE, kind)
+            info["Config" if resource == "container" else "Labels"] = {"Labels": labels} if resource == "container" else labels
+            objects[(resource, name)] = info
+        return objects
+
+    def test_all_ownership_is_verified_before_any_delete(self):
+        for tamper in ("owner", "project", "source_revision", "kind"):
+            objects = self.owned()
+            objects[("network", PROJECT)]["Labels"][repro.LABEL_PREFIX + tamper] = "foreign"
+            docker = FakeDocker(objects)
+            with self.subTest(tamper=tamper), self.assertRaises(repro.ReproError):
+                repro.cleanup_resources(docker, {"project": PROJECT, "source_revision": SOURCE})
+            self.assertEqual(docker.mutations, [], "a later foreign resource must prevent earlier owned deletions")
+
+    def test_cleanup_uses_verified_ids_and_never_wildcards_or_foreign_lists(self):
+        docker = FakeDocker(self.owned())
+        repro.cleanup_resources(docker, {"project": PROJECT, "source_revision": SOURCE,
+                                        "resources": ["malicious-unrelated-container"]})
+        self.assertEqual(len(docker.mutations), 17)
+        container_deletes = docker.mutations[:6]
+        self.assertTrue(all(call[:3] == ("container", "rm", "--force") for call in container_deletes))
+        self.assertTrue(all(call[-1].endswith("-immutable-id") for call in container_deletes))
+        self.assertEqual(docker.mutations[-1], ("network", "rm", PROJECT + "-immutable-id"))
+        self.assertNotIn("malicious-unrelated-container", str(docker.mutations))
+
+    def test_name_mismatch_is_rejected_even_with_correct_labels(self):
+        objects = self.owned()
+        objects[("container", PROJECT + "-init")]["Name"] = "/unrelated"
+        docker = FakeDocker(objects)
+        with self.assertRaises(repro.ReproError):
+            repro.cleanup_resources(docker, {"project": PROJECT, "source_revision": SOURCE})
+        self.assertEqual(docker.mutations, [])
+
+
+class RequestsAndOracles(unittest.TestCase):
+    def test_every_volume_is_verified_before_staging_can_touch_it(self):
+        class StagingDocker(FakeDocker):
+            def __init__(self):
+                super().__init__()
+                self.volume_checks = []
+
+            def inspect(self, resource, name, **kwargs):
+                if resource == "volume":
+                    self.volume_checks.append(name)
+                return super().inspect(resource, name, **kwargs)
+
+            def docker(self, *args, **kwargs):
+                self.mutations.append(args)
+                if args[:2] == ("volume", "create"):
+                    name = args[-1]
+                    labels = dict(args[i + 1].split("=", 1) for i, value in enumerate(args[:-1]) if value == "--label")
+                    # Docker reuses existing names without replacing labels.
+                    self.objects.setdefault(("volume", name), {"Name": name, "Labels": labels})
+                if args[:2] == ("container", "create"):
+                    raise repro.ReproError("stop test at verified staging boundary")
+                return subprocess.CompletedProcess(args, 0, b"", b"")
+
+        for foreign in (False, True):
+            with tempfile.TemporaryDirectory() as root, self.subTest(foreign=foreign):
+                commands = StagingDocker()
+                value = experiment(Path(root) / "case", commands)
+                value.identity = {"id": IMAGE, "architecture": "arm64", "binary_sha256": BINARY}
+                commands.objects[("network", value.project)] = {
+                    "Name": value.project, "Id": "network-id", "Driver": "bridge", "Internal": False,
+                    "Labels": repro.resource_labels(value.project, SOURCE, "network")}
+                collision = value.project + "-config03"
+                if foreign:
+                    commands.objects[("volume", collision)] = {"Name": collision, "Labels": {"owner": "other"}}
+                with patch.object(value, "make_configuration", return_value=b"private-archive"):
+                    expected = "unowned volume" if foreign else "verified staging boundary"
+                    with self.assertRaisesRegex(repro.ReproError, expected):
+                        value.setup()
+                stage = [call for call in commands.mutations if call[:2] == ("container", "create")]
+                if foreign:
+                    self.assertEqual(stage, [], "no init container may mount or chown a foreign volume")
+                    self.assertEqual(commands.objects[("volume", collision)]["Labels"], {"owner": "other"})
+                else:
+                    self.assertEqual(len(commands.volume_checks), 10)
+                    self.assertEqual(len(stage), 1)
+                    mounts = [stage[0][i + 1] for i, value in enumerate(stage[0][:-1]) if value == "--mount"]
+                    self.assertEqual(len(mounts), 10)
+                    self.assertTrue(all("volume-nocopy" in mount for mount in mounts))
+
+    def test_setup_declares_and_records_the_owned_non_internal_bridge(self):
+        class NetworkDocker(FakeDocker):
+            def docker(self, *args, **kwargs):
+                self.mutations.append(args)
+                if args[:2] == ("volume", "create"):
+                    raise repro.ReproError("stop test after network preflight")
+                return subprocess.CompletedProcess(args, 0, b"", b"")
+
+        for driver, internal in (("bridge", False), ("bridge", True), ("overlay", False)):
+            with tempfile.TemporaryDirectory() as root, self.subTest(driver=driver, internal=internal):
+                commands = NetworkDocker()
+                value = experiment(Path(root) / "case", commands)
+                commands.objects[("network", value.project)] = {
+                    "Name": value.project, "Id": "owned-network-id", "Driver": driver, "Internal": internal,
+                    "Labels": repro.resource_labels(value.project, SOURCE, "network")}
+                with patch.object(value, "make_configuration", return_value=b"private-archive"):
+                    expected = "stop test after network preflight" if driver == "bridge" and internal is False else "non-internal bridge"
+                    with self.assertRaisesRegex(repro.ReproError, expected):
+                        value.setup()
+                create = commands.mutations[0]
+                self.assertEqual(create[:4], ("network", "create", "--driver", "bridge"))
+                self.assertNotIn("--internal", create)
+                self.assertEqual(value.summary["network"]["Driver"], driver)
+                self.assertIs(value.summary["network"]["Internal"], internal)
+                if internal or driver != "bridge":
+                    self.assertEqual(len(commands.mutations), 1, "a mismatched network must not create volumes or nodes")
+
+    def test_payload_modes_distinguish_acknowledged_generations_with_identical_size(self):
+        original = [repro.payload_for("original", n) for n in (0, 1, 2)]
+        distinct = [repro.payload_for("distinct", n) for n in (0, 1, 2)]
+        self.assertEqual(len(set(original)), 1)
+        self.assertEqual(len(set(distinct)), 3)
+        self.assertEqual({len(p) for p in original + distinct}, {len(repro.ORIGINAL_PAYLOAD)})
+        self.assertNotEqual(repro.payload_for("distinct", 0, True), distinct[0])
+
+    def test_request_uses_private_sigv4_configuration_and_preserves_binary_body(self):
+        body = b"\xffbinary\x00payload"
+        commands = FakeCurl(body)
+        with tempfile.TemporaryDirectory() as root:
+            value = experiment(Path(root) / "case", commands)
+            record = value.request(2, "GET", "/repro/reference-object")
+            self.assertEqual(record["body_sha256"], repro.sha256(body))
+            self.assertEqual((value.out / record["body"]).read_bytes(), body)
+            self.assertNotIn(value.access, str(commands.calls))
+            self.assertNotIn(value.secret, str(commands.calls))
+            self.assertEqual(commands.calls, [["curl", "--disable", "--config", str(value.out / "private/request-00001.curl")]])
+            self.assertIn('aws-sigv4 = "aws:amz:us-east-1:s3"', commands.configs[0])
+            self.assertIn('noproxy = "*"', commands.configs[0])
+            self.assertIn("retry = 0", commands.configs[0])
+            self.assertIn("storage02:19002:127.0.0.1", commands.configs[0])
+            self.assertEqual(record["generation"]["started_at"], "generation-one")
+
+    def test_responses_redact_credentials_and_classify_startup_or_lock_failures(self):
+        commands = FakeCurl(b"test-secret-private <Code>ServiceUnavailable</Code>", 503,
+                            "HTTP/1.1 503\r\nRetry-After: 1\r\n")
+        with tempfile.TemporaryDirectory() as root:
+            value = experiment(Path(root) / "case", commands)
+            record = value.request(1, "GET", "/repro/degraded-object")
+            self.assertEqual(record["cause"]["retry-after"], "1")
+            self.assertEqual(record["cause"]["s3_code"], "ServiceUnavailable")
+            self.assertNotIn(value.secret, (value.out / record["body"]).read_text())
+            self.assertNotIn(value.secret, (value.out / "events.jsonl").read_text())
+        cause = repro.response_cause(503, "X-Rustfs-Readiness-Pending: startup_finalization\r\n", b"pending")
+        self.assertEqual(cause["classification"], "startup_admission")
+        unavailable = repro.response_cause(503, "X-Rustfs-Error-Cause: namespace_lock_unavailable\r\n", b"unavailable")
+        self.assertEqual(unavailable["x-rustfs-error-cause"], "namespace_lock_unavailable")
+        self.assertNotIn("retry-after", unavailable)
+
+    def test_wrong_overwrite_is_a_failure_and_later_success_does_not_clear_it(self):
+        commands = FakeCurl(repro.payload_for("distinct", 1))
+        with tempfile.TemporaryDirectory() as root:
+            value = experiment(Path(root) / "case", commands)
+            record = value.request(3, "GET", "/repro/degraded-object")
+            with self.assertRaises(repro.ReproError):
+                value.require(record, "degraded", repro.payload_for("distinct", 2))
+            first = value.first_failure
+            value.require(record, "later-convergence", repro.payload_for("distinct", 1))
+            self.assertIs(value.first_failure, first)
+
+    def test_status_parser_and_curl_config_reject_ambiguous_input(self):
+        self.assertEqual(repro.parse_http_status(b"200\n"), 200)
+        self.assertEqual(repro.parse_http_status(b"000"), 0)
+        for output in (b"200200", b"HTTP 200", b"", b"20"):
+            with self.subTest(output=output), self.assertRaises(repro.ReproError):
+                repro.parse_http_status(output)
+        for value in ("path\nuser=leak", "bad\rline", "nul\0"):
+            with self.subTest(value=value), self.assertRaises(repro.ReproError):
+                repro.curl_quote(value)
+        self.assertEqual(repro.curl_quote('a\\b"c'), '"a\\\\b\\"c"')
+
+    def test_failure_always_recovers_and_captures_but_never_cleans(self):
+        with tempfile.TemporaryDirectory() as root:
+            value = experiment(Path(root) / "case", FakeDocker())
+            value.args.cleanup_on_success = True
+            order = []
+
+            def sequence():
+                order.append("sequence")
+                value.fail("degraded", "retained-first-failure")
+                raise repro.ReproError("stop immediately")
+
+            with patch.object(value, "preflight"), patch.object(value, "setup", side_effect=lambda: order.append("setup")), \
+                 patch.object(value, "sequence", side_effect=sequence), \
+                 patch.object(value, "recover", side_effect=lambda: order.append("recover")), \
+                 patch.object(value, "capture", side_effect=lambda: order.append("capture")), \
+                 patch.object(value, "stop_retained_nodes", side_effect=lambda **kwargs: order.append("stop")), \
+                 patch.object(repro, "cleanup_resources") as cleanup, contextlib.redirect_stdout(io.StringIO()):
+                self.assertEqual(value.run(), 2)
+                cleanup.assert_not_called()
+            self.assertEqual(order, ["setup", "sequence", "recover", "capture", "stop"])
+            summary = json.loads((value.out / "summary.json").read_text())
+            self.assertEqual(summary["first_failure"]["reason"], "retained-first-failure")
+            self.assertTrue(summary["resources_retained"])
+
+    def test_retention_validates_all_nodes_then_stops_only_owned_running_ids(self):
+        with tempfile.TemporaryDirectory() as root:
+            commands = FakeDocker()
+            value = experiment(Path(root) / "case", commands)
+            for n, node in value.nodes.items():
+                commands.objects[("container", node["id"])] = {
+                    "Id": node["id"], "Name": "/" + node["name"], "State": {"Running": n != 5},
+                    "Config": {"Labels": repro.resource_labels(value.project, SOURCE, "node")}}
+
+            value.stop_retained_nodes()
+            self.assertEqual(commands.mutations, [("container", "stop", "-t", "10", *(str(n) * 64 for n in range(1, 5)))])
+            snapshot = json.loads((value.out / "containers-retained-stopped.json").read_text())
+            self.assertEqual(len(snapshot), 5)
+            self.assertTrue(all(not item["state"]["Running"] for item in snapshot))
+            self.assertTrue(value.summary["retained_nodes_stopped"])
+            commands.mutations.clear()
+            foreign = commands.objects[("container", value.nodes[5]["id"])]
+            foreign["Config"]["Labels"][repro.LABEL_PREFIX + "project"] = "other-experiment"
+            with self.assertRaises(repro.ReproError):
+                value.stop_retained_nodes()
+            self.assertEqual(commands.mutations, [], "a later foreign node must prevent every stop")
+            del commands.objects[("container", value.nodes[1]["id"])]
+            with self.assertRaises(repro.ReproError):
+                value.stop_retained_nodes(allow_missing=True)
+            self.assertEqual(commands.mutations, [], "partial cleanup must never permit a foreign node to be ignored")
+
+    def test_stop_or_capture_failure_is_reported_and_cannot_turn_into_success(self):
+        with tempfile.TemporaryDirectory() as root:
+            value = experiment(Path(root) / "case", FakeDocker())
+            with patch.object(value, "preflight"), patch.object(value, "setup"), patch.object(value, "sequence"), \
+                 patch.object(value, "recover"), patch.object(value, "capture", side_effect=repro.ReproError("capture failed")), \
+                 patch.object(value, "stop_retained_nodes", side_effect=repro.ReproError("stop failed")), \
+                 contextlib.redirect_stdout(io.StringIO()):
+                self.assertEqual(value.run(), 2)
+            summary = json.loads((value.out / "summary.json").read_text())
+            self.assertEqual(summary["first_failure"]["phase"], "capture")
+            self.assertFalse(summary["retained_nodes_stopped"])
+            events = (value.out / "events.jsonl").read_text()
+            self.assertIn("stop failed", events)
+
+    def test_final_capture_interrupt_still_stops_owned_nodes_and_writes_failed_summary(self):
+        with tempfile.TemporaryDirectory() as root:
+            commands = FakeDocker()
+            value = experiment(Path(root) / "case", commands)
+            for node in value.nodes.values():
+                commands.objects[("container", node["id"])] = {
+                    "Id": node["id"], "Name": "/" + node["name"], "State": {"Running": True},
+                    "Config": {"Labels": repro.resource_labels(value.project, SOURCE, "node")}}
+            with patch.object(value, "preflight"), patch.object(value, "setup"), patch.object(value, "sequence"), \
+                 patch.object(value, "recover"), patch.object(value, "capture", side_effect=KeyboardInterrupt), \
+                 contextlib.redirect_stdout(io.StringIO()):
+                self.assertEqual(value.run(), 2)
+            self.assertEqual(commands.mutations, [("container", "stop", "-t", "10", *(str(n) * 64 for n in range(1, 6)))])
+            summary = json.loads((value.out / "summary.json").read_text())
+            self.assertEqual(summary["first_failure"]["phase"], "capture")
+            self.assertEqual(summary["first_failure"]["reason"], "interrupted")
+            self.assertTrue(summary["retained_nodes_stopped"])
+
+    def test_cleanup_interrupt_falls_back_to_stopping_remaining_owned_nodes(self):
+        with tempfile.TemporaryDirectory() as root:
+            commands = FakeDocker()
+            value = experiment(Path(root) / "case", commands)
+            value.args.cleanup_on_success = True
+            repro.write_json(value.out / "manifest.json", {"project": value.project, "source_revision": SOURCE})
+            for node in value.nodes.values():
+                commands.objects[("container", node["id"])] = {
+                    "Id": node["id"], "Name": "/" + node["name"], "State": {"Running": True},
+                    "Config": {"Labels": repro.resource_labels(value.project, SOURCE, "node")}}
+
+            def interrupted_cleanup(*args):
+                del commands.objects[("container", value.nodes[1]["id"])]
+                raise KeyboardInterrupt
+
+            with patch.object(value, "preflight"), patch.object(value, "setup"), patch.object(value, "sequence"), \
+                 patch.object(value, "recover"), patch.object(value, "capture"), \
+                 patch.object(repro, "cleanup_resources", side_effect=interrupted_cleanup), \
+                 contextlib.redirect_stdout(io.StringIO()):
+                self.assertEqual(value.run(), 2)
+            self.assertEqual(commands.mutations, [("container", "stop", "-t", "10", *(str(n) * 64 for n in range(2, 6)))])
+            summary = json.loads((value.out / "summary.json").read_text())
+            self.assertEqual(summary["first_failure"]["phase"], "cleanup")
+            self.assertEqual(summary["first_failure"]["reason"], "interrupted")
+            self.assertTrue(summary["retained_nodes_stopped"])
+
+    def test_retention_interrupt_is_recorded_in_the_failed_summary(self):
+        with tempfile.TemporaryDirectory() as root:
+            value = experiment(Path(root) / "case", FakeDocker())
+            with patch.object(value, "preflight"), patch.object(value, "setup"), patch.object(value, "sequence"), \
+                 patch.object(value, "recover"), patch.object(value, "capture"), \
+                 patch.object(value, "stop_retained_nodes", side_effect=KeyboardInterrupt), \
+                 contextlib.redirect_stdout(io.StringIO()):
+                self.assertEqual(value.run(), 2)
+            summary = json.loads((value.out / "summary.json").read_text())
+            self.assertEqual(summary["first_failure"]["phase"], "retention-stop")
+            self.assertFalse(summary["retained_nodes_stopped"])
+
+    def test_partial_cleanup_failure_stops_remaining_owned_nodes_and_retains_the_error(self):
+        with tempfile.TemporaryDirectory() as root:
+            commands = FakeDocker()
+            value = experiment(Path(root) / "case", commands)
+            value.args.cleanup_on_success = True
+            repro.write_json(value.out / "manifest.json", {"project": value.project, "source_revision": SOURCE})
+            for n, node in value.nodes.items():
+                if n == 1:
+                    continue  # cleanup removed this node before failing
+                commands.objects[("container", node["id"])] = {
+                    "Id": node["id"], "Name": "/" + node["name"], "State": {"Running": True},
+                    "Config": {"Labels": repro.resource_labels(value.project, SOURCE, "node")}}
+
+            with patch.object(value, "preflight"), patch.object(value, "setup"), patch.object(value, "sequence"), \
+                 patch.object(value, "recover"), patch.object(value, "capture"), \
+                 patch.object(repro, "cleanup_resources", side_effect=repro.ReproError("partial cleanup failure")), \
+                 contextlib.redirect_stdout(io.StringIO()):
+                self.assertEqual(value.run(), 2)
+            self.assertEqual(commands.mutations, [("container", "stop", "-t", "10", *(str(n) * 64 for n in range(2, 6)))])
+            summary = json.loads((value.out / "summary.json").read_text())
+            self.assertEqual(summary["first_failure"]["reason"], "partial cleanup failure")
+            self.assertTrue(summary["retained_nodes_stopped"])
+            self.assertTrue(summary["resources_retained"])
+
+    def test_staging_timeout_with_no_recorded_nodes_stops_the_owned_init_container(self):
+        with tempfile.TemporaryDirectory() as root:
+            commands = FakeDocker()
+            value = experiment(Path(root) / "case", commands)
+            value.nodes = {}
+            name, identifier = value.project + "-init", "f" * 64
+            commands.objects[("container", name)] = {"Id": identifier, "Name": "/" + name, "State": {"Running": True},
+                "Config": {"Labels": repro.resource_labels(value.project, SOURCE, "init")}}
+            with patch.object(value, "preflight"), patch.object(value, "setup", side_effect=repro.ReproError("staging timeout")), \
+                 contextlib.redirect_stdout(io.StringIO()):
+                self.assertEqual(value.run(), 2)
+            self.assertEqual(commands.mutations, [("container", "stop", "-t", "10", identifier)])
+            self.assertEqual(value.summary["first_failure"]["reason"], "staging timeout")
+            self.assertTrue(value.summary["retained_nodes_stopped"])
+            snapshot = json.loads((value.out / "containers-retained-stopped.json").read_text())
+            self.assertEqual([(item["kind"], item["state"]["Running"]) for item in snapshot], [("init", False)])
+            commands.mutations.clear()
+            commands.objects[("container", name)]["Config"]["Labels"][repro.LABEL_PREFIX + "owner"] = "foreign"
+            with self.assertRaises(repro.ReproError):
+                value.stop_retained_nodes()
+            self.assertEqual(commands.mutations, [], "a foreign init container must prevent every mutation")
+
+    def test_missing_resource_is_distinguished_from_transport_and_permission_errors(self):
+        commands = repro.Commands()
+        missing = {"container": ("Error: No such container: gone", "Error response from daemon: No such container: gone"),
+                   "volume": ("Error response from daemon: get gone: no such volume",),
+                   "network": ("Error response from daemon: network gone not found",)}
+        for resource, errors in missing.items():
+            for error in errors:
+                with self.subTest(resource=resource, error=error), patch.object(commands, "docker", return_value=subprocess.CompletedProcess([], 1, b"[]", error.encode())):
+                    self.assertIsNone(commands.inspect(resource, "gone", optional=True))
+            for error in ("Cannot connect to the Docker daemon: no such file or directory", "permission denied", "credential helper not found",
+                          errors[0] + "\nCannot connect to Docker daemon", errors[0].replace("gone", "other-resource")):
+                with self.subTest(resource=resource, error=error), patch.object(commands, "docker", return_value=subprocess.CompletedProcess([], 1, b"[]", error.encode())):
+                    with self.assertRaises(repro.ReproError):
+                        commands.inspect(resource, "gone", optional=True)
+
+    def test_recovery_gate_is_followed_by_object_reads_before_classification(self):
+        with tempfile.TemporaryDirectory() as root:
+            value = experiment(Path(root) / "case", FakeDocker())
+            value.started = True
+            value.acknowledged = {"reference-object": b"ref", "degraded-object": b"last"}
+            order = []
+            for node in value.nodes.values():
+                node["running"] = True
+
+            def request(node, method, path, **kwargs):
+                order.append((kwargs["phase"], node, path))
+                key = path.rsplit("/", 1)[-1]
+                return {"id": len(order), "curl_exit": 0, "http": 200,
+                        "body_sha256": repro.sha256(value.acknowledged.get(key, b""))}
+
+            def gate(phase):
+                order.append(("gate", phase))
+                return {"monotonic": 0}
+
+            with patch.object(value, "snapshot"), patch.object(value, "request", side_effect=request), \
+                 patch.object(value, "wait_gate", side_effect=gate):
+                value.recover()
+            self.assertEqual(order[0], ("gate", "recovery"))
+            self.assertTrue(all(entry[0] == "immediate-recovery" for entry in order[1:11]))
+            self.assertTrue(all(entry[0] == "post-recovery-classification" for entry in order[11:]))
+            self.assertEqual(value.summary["recovery_gets"]["count"], 10)
+
+    def test_gate_override_checks_and_records_node_write_without_changing_the_case(self):
+        with tempfile.TemporaryDirectory() as root:
+            value = experiment(Path(root) / "case", FakeDocker())
+            calls = []
+
+            def request(node, method, path, **kwargs):
+                calls.append((node, path))
+                return {"http": 200, "curl_exit": 0}
+
+            with patch.object(value, "request", side_effect=request):
+                gate = value.wait_gate("convergence", gate="node-write")
+            self.assertEqual(value.args.gate, "cluster")
+            self.assertEqual(gate["gate"], "node-write")
+            self.assertEqual(gate["paths"], ["/health/ready", "/minio/health/cluster"])
+            self.assertEqual(calls, [(node, path) for node in range(1, 6) for path in repro.GATE_PATHS["node-write"]])
+
+    def test_later_success_is_separate_and_cannot_clear_immediate_failure_or_exit_status(self):
+        with tempfile.TemporaryDirectory() as root:
+            value = experiment(Path(root) / "case", FakeDocker())
+            value.started = True
+            value.acknowledged = {"reference-object": b"ref", "degraded-object": b"last"}
+            for node in value.nodes.values():
+                node["running"] = True
+            requests, gates = [], []
+
+            def gate(phase, *, gate=None):
+                selected = value.args.gate if gate is None else gate
+                gates.append((phase, selected))
+                return {"gate": selected, "paths": list(repro.GATE_PATHS[selected]), "monotonic": time.monotonic(),
+                        "completed_at": repro.utc_now()}
+
+            def request(node, method, path, **kwargs):
+                phase = kwargs["phase"]
+                requests.append(phase)
+                key = path.rsplit("/", 1)[-1]
+                failed = phase == "immediate-recovery" and node == 2 and key == "reference-object"
+                return {"id": len(requests), "phase": phase, "node": node, "http": 503 if failed else 200,
+                        "curl_exit": 0, "body_sha256": repro.sha256(value.acknowledged.get(key, b""))}
+
+            with patch.object(value, "preflight"), patch.object(value, "setup"), patch.object(value, "sequence"), \
+                 patch.object(value, "snapshot"), patch.object(value, "capture"), patch.object(value, "stop_retained_nodes"), \
+                 patch.object(value, "wait_gate", side_effect=gate), patch.object(value, "request", side_effect=request), \
+                 contextlib.redirect_stdout(io.StringIO()):
+                self.assertEqual(value.run(), 2)
+            self.assertEqual(gates, [("recovery", "cluster"), ("convergence", "node-write")])
+            self.assertEqual(value.args.gate, "cluster")
+            self.assertEqual(requests, ["immediate-recovery"] * 10 + ["post-recovery-classification"] * 15 + ["later-convergence"] * 10)
+            self.assertEqual(value.summary["result"], "failed")
+            self.assertEqual(value.summary["first_failure"]["phase"], "immediate-recovery")
+            self.assertEqual(value.summary["recovery_gets"]["failed_request_ids"], [2])
+            self.assertEqual(value.summary["convergence"]["gate"]["gate"], "node-write")
+            self.assertEqual(value.summary["convergence"]["gets"]["count"], 10)
+            self.assertEqual(value.summary["convergence"]["result"], "passed")
+            self.assertIn("started_at", value.summary["convergence"])
+            self.assertIn("finished_at", value.summary["convergence"])
+
+    def test_lock_failure_snapshots_running_generation_before_any_restart(self):
+        with tempfile.TemporaryDirectory() as root:
+            commands = FakeDocker()
+            value = experiment(Path(root) / "case", commands)
+            value.started = True
+            value.args.lock_samples = 1
+            value.first_failure = {"request": {"cause": {"x-rustfs-error-cause": "namespace_lock_unavailable"}}}
+            value.nodes[5]["running"] = False
+            for node in range(1, 5):
+                value.nodes[node]["running"] = True
+            order = []
+
+            def request(node, method, path, **kwargs):
+                order.append((kwargs["phase"], node))
+                self.assertEqual(commands.mutations, [], "lock evidence must precede recovery mutation")
+                return {"id": node, "http": 404, "curl_exit": 0}
+
+            def gate(phase):
+                order.append(("gate", phase))
+                raise repro.ReproError("stop test after the restart")
+
+            with patch.object(value, "snapshot"), patch.object(value, "capture", side_effect=lambda phase: order.append(("capture", phase))), \
+                 patch.object(value, "request", side_effect=request), patch.object(value, "wait_gate", side_effect=gate):
+                with self.assertRaises(repro.ReproError):
+                    value.recover()
+            self.assertEqual({node for phase, node in order if phase.startswith("pre-recovery-locks")}, {1, 2, 3, 4})
+            self.assertEqual(commands.mutations, [("container", "start", "5" * 64)])
+            self.assertTrue(all(phase == "pre-recovery-locks-1" for phase, _ in order[:4]))
+            self.assertEqual(order[4], ("capture", "before-recovery"), "volatile locks must be captured before file logs")
+            self.assertEqual(value.first_failure["request"]["cause"]["x-rustfs-error-cause"], "namespace_lock_unavailable")
+
+
+if __name__ == "__main__":
+    unittest.main()