ci: prepare scoped TLS resource observation

.github/scripts/observe_tls_resource.py

@@ -0,0 +1,466 @@
+#!/usr/bin/env python3
+"""Collect bounded, allowlisted local metrics for one fixed TLS e2e case."""
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import signal
+import subprocess
+import time
+import xml.etree.ElementTree as ET
+
+SOURCE = "fedc9666123076f2b985d75a9c290f6496ef8717"
+TREE = "1ea486f6d6a86ce488b83cab4a1b232188ab9708"
+CASE = "distributed::internode_tls_test::four_node_tls_large_and_range_gets_preserve_bytes"
+FILTER = f"test(={CASE})"
+INTERVAL_SECONDS = 2
+MAX_SAMPLES = 900
+MAX_SCAN_PIDS = 2048
+SCAN_EVERY_SAMPLES = 5
+MAX_PROCESS_SCANS = 120
+MAX_ANCESTRY_DEPTH = 32
+PRIVATE = Path(os.environ["RUNNER_TEMP"]) / "tls-private"
+EVIDENCE = Path(os.environ["RUNNER_TEMP"]) / "tls-evidence"
+REPORT = PRIVATE / "acceptance.json"
+PIDS = PRIVATE / "acceptance.pids.json"
+JUNIT = Path("target/nextest/e2e-distributed/junit.xml")
+TEST_LOG = PRIVATE / "test.log"
+
+
+def digest(path):
+    h = hashlib.sha256()
+    with open(path, "rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def write_json(name, value):
+    (EVIDENCE / name).write_text(json.dumps(value, sort_keys=True, allow_nan=False) + "\n")
+
+
+def read_text(path):
+    try:
+        return Path(path).read_text(errors="replace"), "available"
+    except PermissionError:
+        return None, "denied"
+    except FileNotFoundError:
+        return None, "missing"
+    except OSError:
+        return None, "unavailable"
+
+
+def process_stat(pid):
+    raw, state = read_text(f"/proc/{pid}/stat")
+    if raw is None:
+        return None, state
+    try:
+        # comm can contain spaces and parentheses; no name or command is retained.
+        fields = raw[raw.rfind(")") + 2 :].split()
+        data = {
+            "state_code": fields[0],
+            "ppid": int(fields[1]),
+            "user_ticks": int(fields[11]),
+            "system_ticks": int(fields[12]),
+            "start_ticks": int(fields[19]),
+            "rss_pages": int(fields[21]),
+        }
+        if data["start_ticks"] < 0 or data["rss_pages"] < 0:
+            raise ValueError
+        return data, "available"
+    except (IndexError, ValueError):
+        return None, "malformed"
+
+
+def process_io(pid):
+    raw, state = read_text(f"/proc/{pid}/io")
+    if raw is None:
+        return {"state": state, "values": None}
+    wanted = {"rchar", "wchar", "read_bytes", "write_bytes", "cancelled_write_bytes"}
+    values = {}
+    try:
+        for line in raw.splitlines():
+            key, value = line.split(":", 1)
+            if key in wanted:
+                values[key] = int(value.strip())
+        if set(values) != wanted or any(v < 0 for v in values.values()):
+            raise ValueError
+        return {"state": "available", "values": values}
+    except ValueError:
+        return {"state": "malformed", "values": None}
+
+
+def sample_process(identity):
+    if identity is None:
+        return {"state": "not_observed", "pid": None, "start_ticks": None, "cpu": None, "rss_bytes": None, "io": None}
+    pid, expected_start = identity
+    stat, state = process_stat(pid)
+    if stat is None:
+        return {"state": state, "pid": pid, "start_ticks": expected_start, "cpu": None, "rss_bytes": None, "io": None}
+    if stat["start_ticks"] != expected_start:
+        return {"state": "pid_reused", "pid": pid, "start_ticks": expected_start, "cpu": None, "rss_bytes": None, "io": None}
+    io = process_io(pid)
+    again, after_state = process_stat(pid)
+    if again is None or again["start_ticks"] != expected_start:
+        return {"state": "raced_" + after_state, "pid": pid, "start_ticks": expected_start, "cpu": None, "rss_bytes": None, "io": None}
+    state = "zombie" if stat["state_code"] in ("Z", "X") else "alive"
+    return {
+        "state": state,
+        "pid": pid,
+        "start_ticks": expected_start,
+        "cpu": {"user_ticks": stat["user_ticks"], "system_ticks": stat["system_ticks"]},
+        "rss_bytes": stat["rss_pages"] * os.sysconf("SC_PAGE_SIZE"),
+        "io": io,
+    }
+
+
+def executable_matches(pid, executable):
+    try:
+        actual = os.stat(f"/proc/{pid}/exe")
+        return (actual.st_dev, actual.st_ino) == executable
+    except OSError:
+        return False
+
+
+def process_tree(root):
+    try:
+        pids = [int(name) for name in os.listdir("/proc") if name.isdigit()]
+    except OSError:
+        return {}, "unavailable"
+    if len(pids) > MAX_SCAN_PIDS:
+        return {}, "too_many_processes"
+    children_by_parent = {}
+    for pid in pids:
+        stat, _ = process_stat(pid)
+        if stat is not None:
+            children_by_parent.setdefault(stat["ppid"], []).append(pid)
+    children, pending = set(), [root]
+    while pending:
+        current = pending.pop()
+        if current not in children:
+            children.add(current)
+            pending.extend(children_by_parent.get(current, []))
+    return children, "available"
+
+
+def find_test(root, descendants, executable, scan_state):
+    candidates = []
+    for pid in descendants:
+        before, _ = process_stat(pid)
+        if before is not None and executable_matches(pid, executable) and is_descendant(pid, root):
+            after, _ = process_stat(pid)
+            if after is not None and before["start_ticks"] == after["start_ticks"]:
+                candidates.append((pid, after["start_ticks"]))
+    if len(candidates) == 1:
+        return candidates[0], scan_state
+    return None, "ambiguous" if len(candidates) > 1 else scan_state
+
+
+def is_descendant(pid, root):
+    seen = set()
+    for _ in range(MAX_ANCESTRY_DEPTH):
+        if pid == root:
+            return True
+        if pid in seen:
+            return False
+        seen.add(pid)
+        stat, _ = process_stat(pid)
+        if stat is None:
+            return False
+        pid = stat["ppid"]
+    return False
+
+
+def identify_nodes(node_ids, root, executable, known):
+    if node_ids is None:
+        raw, _ = read_text(PIDS)
+        try:
+            candidate = json.loads(raw) if raw is not None else None
+            if (isinstance(candidate, list) and len(candidate) == 4
+                    and all(type(pid) is int and pid > 0 for pid in candidate)
+                    and len(set(candidate)) == 4):
+                node_ids = candidate
+        except (ValueError, TypeError):
+            pass
+    if node_ids is None:
+        return None, known, "not_observed"
+    if all(identity is not None for identity in known):
+        return node_ids, known, "available"
+    for index, pid in enumerate(node_ids):
+        if known[index] is None:
+            before, _ = process_stat(pid)
+            if before is not None and executable_matches(pid, executable) and is_descendant(pid, root):
+                after, _ = process_stat(pid)
+                if after is not None and after["start_ticks"] == before["start_ticks"]:
+                    known[index] = (pid, after["start_ticks"])
+    return node_ids, known, "available" if all(identity is not None for identity in known) else "partial"
+
+
+def host_cpu():
+    raw, state = read_text("/proc/stat")
+    if raw is None:
+        return {"state": state, "values": None}
+    try:
+        line = next(line for line in raw.splitlines() if line.startswith("cpu "))
+        numbers = [int(value) for value in line.split()[1:9]]
+        if len(numbers) != 8 or any(value < 0 for value in numbers):
+            raise ValueError
+        return {"state": "available", "values": dict(zip(("user", "nice", "system", "idle", "iowait", "irq", "softirq", "steal"), numbers))}
+    except (StopIteration, ValueError):
+        return {"state": "malformed", "values": None}
+
+
+def host_io():
+    raw, state = read_text("/proc/vmstat")
+    if raw is None:
+        return {"state": state, "values": None}
+    wanted = {"pgpgin", "pgpgout", "pswpin", "pswpout"}
+    try:
+        values = {key: int(value) for line in raw.splitlines() for key, value in [line.split()] if key in wanted}
+        if set(values) != wanted or any(value < 0 for value in values.values()):
+            raise ValueError
+        return {"state": "available", "values": values}
+    except ValueError:
+        return {"state": "malformed", "values": None}
+
+
+def pressure(resource):
+    raw, state = read_text(f"/proc/pressure/{resource}")
+    if raw is None:
+        return {"state": state, "values": None}
+    try:
+        values = {}
+        for line in raw.splitlines():
+            parts = line.split()
+            if parts[0] not in ("some", "full"):
+                continue
+            fields = dict(part.split("=", 1) for part in parts[1:])
+            avg10 = float(fields["avg10"])
+            total = int(fields["total"])
+            if not (0 <= avg10 <= 100 and total >= 0):
+                raise ValueError
+            values[parts[0]] = {"avg10_percent": avg10, "total_us": total}
+        if "some" not in values:
+            raise ValueError
+        # Some kernels do not provide full CPU pressure.
+        return {"state": "available", "values": {"some": values["some"], "full": values.get("full")}}
+    except (ValueError, KeyError, IndexError):
+        return {"state": "malformed", "values": None}
+
+
+def nullable_number(value):
+    return value if type(value) is int and 0 <= value <= 2**63 - 1 else None
+
+
+def clock_fields(clock):
+    if not isinstance(clock, dict):
+        clock = {}
+    return {key: nullable_number(clock.get(key)) for key in ("unix_ms", "monotonic_before_ms", "monotonic_after_ms")}
+
+
+def progress_evidence():
+    prefix = "TLS acceptance: concurrent GET failure: "
+    snapshots = []
+    if not TEST_LOG.exists():
+        return {"state": "missing_test_output", "snapshots": snapshots}
+    with TEST_LOG.open(errors="replace") as lines:
+        for line in lines:
+            if prefix not in line or len(line) > 100000:
+                continue
+            try:
+                value = json.loads(line.split(prefix, 1)[1])
+                if value.get("kind") not in ("read_failed", "task_failed") or not isinstance(value.get("streams"), list):
+                    continue
+                streams = []
+                for stream in value["streams"][:4]:
+                    if not isinstance(stream, dict):
+                        continue
+                    stage = stream.get("last_stage")
+                    terminal = stream.get("terminal")
+                    snapshot = stream.get("snapshot")
+                    streams.append({
+                        "node": nullable_number(stream.get("node")),
+                        "snapshot": snapshot if snapshot in ("available", "busy", "poisoned") else "unknown",
+                        "stage": stage if stage in ("queued", "waiting_headers", "headers_received", "reading_body", "complete") else "unknown",
+                        "terminal": terminal if terminal in ("success", "failed", "panicked", "cancelled") else "unknown",
+                        "sampled_clock": clock_fields(stream.get("sampled_clock")),
+                        **{key: nullable_number(stream.get(key)) for key in ("started_ms", "headers_ms", "body_started_ms", "completed_ms", "received_bytes", "last_progress_ms", "terminal_ms")},
+                    })
+                snapshots.append({
+                    "kind": value["kind"], "failed_node": nullable_number(value.get("failed_node")),
+                    "elapsed_ms": nullable_number(value.get("elapsed_ms")),
+                    "expected_bytes": nullable_number(value.get("expected_bytes")),
+                    "stage_start_clock": clock_fields(value.get("stage_start_clock_sample")), "streams": streams,
+                })
+            except (ValueError, TypeError, AttributeError):
+                continue
+    return {"state": "available" if snapshots else "not_observed", "snapshots": snapshots[:4]}
+
+
+def sanitize_junit():
+    if not JUNIT.exists():
+        return "missing"
+    try:
+        cases = list(ET.parse(JUNIT).getroot().iter("testcase"))
+        if len(cases) != 1 or cases[0].get("name") != CASE or cases[0].get("classname") != "e2e_test":
+            return "unexpected_case_set"
+        source = cases[0]
+        outcome = "failure" if source.find("failure") is not None else "error" if source.find("error") is not None else "skipped" if source.find("skipped") is not None else "passed"
+        duration = float(source.get("time", ""))
+        if not 0 <= duration < 86400:
+            raise ValueError
+        root = ET.Element("testsuites")
+        suite = ET.SubElement(root, "testsuite", {"name": "e2e_test", "tests": "1", "failures": str(int(outcome == "failure")), "errors": str(int(outcome == "error")), "skipped": str(int(outcome == "skipped")), "time": str(duration)})
+        case = ET.SubElement(suite, "testcase", {"classname": "e2e_test", "name": CASE, "time": str(duration)})
+        if outcome != "passed":
+            ET.SubElement(case, outcome, {"message": "Details withheld"})
+        ET.ElementTree(root).write(EVIDENCE / "junit.xml", encoding="utf-8", xml_declaration=True)
+        return outcome
+    except (ET.ParseError, ValueError, OSError):
+        return "malformed"
+
+
+def listing_and_identity():
+    listing = PRIVATE / "listing.json"
+    with listing.open("wb") as output, (PRIVATE / "list.log").open("wb") as errors:
+        result = subprocess.run(["cargo", "nextest", "list", "--locked", "--profile", "e2e-distributed", "-p", "e2e_test", "-E", FILTER, "--message-format", "json"], stdout=output, stderr=errors, check=False)
+    if result.returncode != 0:
+        raise ValueError("listing_failed")
+    data = json.loads(listing.read_text())
+    matches = [(suite_id, name, test) for suite_id, suite in data["rust-suites"].items() for name, test in suite["testcases"].items() if test.get("filter-match", {}).get("status") == "matches"]
+    if len(matches) != 1 or matches[0][0:2] != ("e2e_test", CASE) or matches[0][2].get("ignored") is not False:
+        raise ValueError("unexpected_selection")
+    suite = data["rust-suites"]["e2e_test"]
+    if suite.get("binary-id") != "e2e_test" or suite.get("package-name") != "e2e_test":
+        raise ValueError("unexpected_binary")
+    binary = Path(suite["binary-path"])
+    test_stat = binary.stat()
+    server = Path("target/debug/rustfs")
+    server_stat = server.stat()
+    sidecar = json.loads(Path("target/debug/rustfs.e2e.json").read_text())
+    if sidecar["source"]["head"] != SOURCE or sidecar["requested_features"] != ["e2e-test-hooks"] or sidecar["profile"] != "debug" or digest(server) != sidecar["binary_sha256"]:
+        raise ValueError("server_identity_mismatch")
+    manifest = {
+        "source_commit": SOURCE, "source_tree": TREE,
+        "source_inventory_sha256": sidecar["source"]["sha256"],
+        "server_binary_sha256": sidecar["binary_sha256"],
+        "server_requested_features": sidecar["requested_features"],
+        "server_resolved_features": sidecar["features"], "server_profile": sidecar["profile"],
+        "test_binary_sha256": digest(binary), "suite": "e2e_test", "test": CASE, "selected_count": 1,
+        "clock_ticks_per_second": os.sysconf("SC_CLK_TCK"), "interval_seconds": INTERVAL_SECONDS,
+        "max_samples": MAX_SAMPLES, "max_process_scans": MAX_PROCESS_SCANS,
+        "process_scan_interval_samples": SCAN_EVERY_SAMPLES, "max_scan_pids": MAX_SCAN_PIDS,
+        "max_ancestry_depth": MAX_ANCESTRY_DEPTH, "workflow_commit": os.environ.get("WORKFLOW_SHA"),
+        "observer_sha256": digest(Path(__file__)),
+        "metric_scope": "host_and_observed_processes",
+        "run_id": nullable_number(int(os.environ["RUN_ID"])), "run_attempt": nullable_number(int(os.environ["RUN_ATTEMPT"])),
+    }
+    return binary, (test_stat.st_dev, test_stat.st_ino), (server_stat.st_dev, server_stat.st_ino), manifest
+
+
+def main():
+    os.umask(0o077)
+    EVIDENCE.mkdir(parents=True, exist_ok=True)
+    summary = {"state": "preparing", "test_exit_code": None, "junit": "not_observed", "samples": 0, "samples_truncated": False}
+    write_json("summary.json", summary)
+    binary, test_exe, server_exe, manifest = listing_and_identity()
+    write_json("identity.json", manifest)
+    if JUNIT.exists() or JUNIT.is_symlink():
+        summary["state"] = "preexisting_junit_preserved"
+        summary["junit"] = "preexisting_preserved"
+        write_json("summary.json", summary)
+        print("Pre-existing JUnit preserved; test not started.")
+        return 1
+    env = dict(os.environ, RUSTFS_INTERNODE_ACCEPTANCE_REPORT=str(REPORT))
+    command = ["python3", "scripts/e2e_binary.py", "run", "--features", "e2e-test-hooks", "--", "cargo", "nextest", "run", "--locked", "--profile", "e2e-distributed", "-p", "e2e_test", "-E", FILTER, "--no-tests=fail", "--retries", "0", "--failure-output", "immediate", "--success-output", "never", "--color", "never", "--no-output-indent"]
+    test_identity = None
+    node_ids, nodes = None, [None] * 4
+    process_scans = 0
+    with TEST_LOG.open("wb") as output, (EVIDENCE / "samples.jsonl").open("w") as samples:
+        process = subprocess.Popen(command, stdout=output, stderr=subprocess.STDOUT, env=env, start_new_session=True)
+        try:
+            # Lower only the observer's priority after the test wrapper starts.
+            try:
+                os.nice(10)
+                summary["observer_nice_adjusted"] = True
+            except OSError:
+                summary["observer_nice_adjusted"] = False
+            while process.poll() is None:
+                sample_started = time.monotonic()
+                if summary["samples"] < MAX_SAMPLES:
+                    before = time.time_ns() // 1_000_000
+                    mono = time.monotonic_ns() // 1_000_000
+                    scan_state = "not_needed"
+                    if test_identity is None:
+                        if process_scans >= MAX_PROCESS_SCANS:
+                            scan_state = "scan_limit"
+                        elif summary["samples"] % SCAN_EVERY_SAMPLES:
+                            scan_state = "rate_limited"
+                        else:
+                            descendants, scan_state = process_tree(process.pid)
+                            process_scans += 1
+                            test_identity, scan_state = find_test(process.pid, descendants, test_exe, scan_state)
+                    node_ids, nodes, node_scan = identify_nodes(node_ids, process.pid, server_exe, nodes)
+                    node_samples = [{"index": index, "reported_pid": node_ids[index] if node_ids is not None else None,
+                                     **sample_process(identity)} for index, identity in enumerate(nodes)]
+                    record = {"unix_before_ms": before, "monotonic_ms": mono, "test_discovery": scan_state, "node_discovery": node_scan,
+                              "test": sample_process(test_identity), "nodes": node_samples,
+                              "host_cpu": host_cpu(), "host_io": host_io(),
+                              "host_psi": {key: pressure(key) for key in ("cpu", "memory", "io")}, "unix_after_ms": time.time_ns() // 1_000_000}
+                    samples.write(json.dumps(record, sort_keys=True, allow_nan=False) + "\n")
+                    samples.flush()
+                    summary["samples"] += 1
+                else:
+                    summary["samples_truncated"] = True
+                try:
+                    delay = max(0.25, INTERVAL_SECONDS - (time.monotonic() - sample_started))
+                    process.wait(timeout=delay)
+                except subprocess.TimeoutExpired:
+                    pass
+        finally:
+            if process.poll() is None:
+                # e2e_binary.py handles SIGINT by stopping its nextest group.
+                try:
+                    os.killpg(process.pid, signal.SIGINT)
+                    process.wait(timeout=10)
+                except (OSError, subprocess.TimeoutExpired):
+                    try:
+                        os.killpg(process.pid, signal.SIGKILL)
+                    except OSError:
+                        pass
+                    process.wait()
+            summary["test_exit_code"] = process.returncode
+            summary["state"] = "processing"
+            write_json("summary.json", summary)
+    summary["test_exit_code"] = process.returncode
+    summary["junit"] = sanitize_junit()
+    summary["test_process_observed"] = test_identity is not None
+    summary["process_scans"] = process_scans
+    summary["node_processes_observed"] = sum(identity is not None for identity in nodes)
+    summary["node_pid_file_observed"] = node_ids is not None
+    summary["test_binary_unchanged"] = digest(binary) == manifest["test_binary_sha256"]
+    summary["state"] = "complete" if process.returncode == 0 and summary["junit"] == "passed" and summary["test_binary_unchanged"] else "failed"
+    write_json("progress.json", progress_evidence())
+    write_json("summary.json", summary)
+    return 0 if summary["state"] == "complete" else 1
+
+
+if __name__ == "__main__":
+    def interrupted(_signal, _frame):
+        raise KeyboardInterrupt
+
+    signal.signal(signal.SIGTERM, interrupted)
+    try:
+        raise SystemExit(main())
+    except (Exception, KeyboardInterrupt):
+        try:
+            prior = json.loads((EVIDENCE / "summary.json").read_text())
+            exit_code = prior.get("test_exit_code")
+            if type(exit_code) is not int or not -255 <= exit_code <= 255:
+                exit_code = None
+            write_json("summary.json", {"state": "observer_error", "test_exit_code": exit_code, "junit": "not_observed"})
+        except (OSError, ValueError, AttributeError):
+            pass
+        print("Observation failed; raw output withheld.")
+        raise SystemExit(1)

.github/workflows/e2e-distributed.yml

@@ -1,87 +1,16 @@
-# Copyright 2024 RustFS Team
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-# 4-node 4-disk distributed e2e lane.
-#
-# Each selected test starts a real localhost cluster via
-# `RustFSTestClusterEnvironment` (4 processes; 4 drives per node unless the
-# case is a two-site 4-node 1-drive pair or a 4-node upgrade). Membership is
-# `[profile.e2e-distributed]` in `.config/nextest.toml`. Storage-sensitive PRs,
-# nightly runs, and manual dispatches all execute the same fail-closed suite.
-# Upgrade cases download the same pinned previous release as e2e-upgrade.yml.
-#
-# Isolated pool filesystems: expand/decommission/rebalance cases require
-# independent `statfs` capacity. This job runs on GitHub-hosted
-# `ubuntu-latest` because the self-hosted `sm-standard-4` ARC pods cannot
-# create filesystems: `mount -o loop` fails with ENOENT (no
-# `/dev/loop-control`), and `mount -t tmpfs` fails with "cannot mount tmpfs
-# read-only" (no `CAP_SYS_ADMIN` in the initial namespace). The same reason
-# `uring-integration` and `e2e-s3tests.yml` left that label. The prepare
-# step mounts four 1 GiB tmpfs instances and exports `RUSTFS_E2E_POOL_ROOTS`.
-
-name: e2e-distributed
+# Temporary, manually dispatched diagnostic for one distributed TLS case.
+# The workflow branch carries only the observer; both builds use the fixed source.
+name: Distributed TLS resource observation
 
 on:
-  pull_request:
-    paths:
-      - "Cargo.lock"
-      - "Cargo.toml"
-      - ".config/nextest.toml"
-      - ".config/e2e-distributed-selection.txt"
-      - ".github/workflows/e2e-distributed.yml"
-      - "crates/audit/**"
-      - "crates/common/**"
-      - "crates/config/**"
-      - "crates/e2e_test/**"
-      - "crates/ecstore/**"
-      - "crates/filemeta/**"
-      - "crates/heal/**"
-      - "crates/iam/**"
-      - "crates/lock/**"
-      - "crates/madmin/**"
-      - "crates/notify/**"
-      - "crates/replication/**"
-      - "crates/s3-client/**"
-      - "crates/s3-ops/**"
-      - "crates/s3-types/**"
-      - "crates/scanner/**"
-      - "crates/storage-api/**"
-      - "crates/utils/**"
-      - "rustfs/**"
   workflow_dispatch:
-    inputs:
-      filter:
-        description: "Optional nextest -E filter (default: the whole e2e-distributed profile)"
-        required: false
-        default: ""
-  schedule:
-    # 05:53 UTC nightly — clear of e2e-nightly (04:29) and ODM interop (05:23).
-    - cron: "53 5 * * *"
 
 permissions:
   contents: read
 
-concurrency:
-  group: ${{ github.workflow }}-${{ github.ref }}
-  cancel-in-progress: ${{ github.event_name != 'schedule' }}
-
 jobs:
-  distributed:
+  observe:
     if: github.repository == 'rustfs/rustfs'
-    name: Distributed 4-node 4-disk e2e
-    # GitHub-hosted VM: loop and tmpfs mounts work here. sm-standard-4 is an
-    # ARC pod and rejects both (`mount -o loop` ENOENT, tmpfs "read-only").
     runs-on: ubuntu-latest
     timeout-minutes: 180
     env:
@@ -89,131 +18,184 @@ jobs:
       NO_PROXY: 127.0.0.1,localhost
       HTTP_PROXY: ""
       HTTPS_PROXY: ""
-      # Pinned previous release used by distributed::upgrade_test (same pin as e2e-upgrade.yml).
-      UPGRADE_SOURCE_VERSION: 1.0.0-rc.2
-      UPGRADE_SOURCE_ASSET: rustfs-linux-x86_64-gnu-v1.0.0-rc.2.zip
-      UPGRADE_SOURCE_SHA256: 7c789386bf85278f865b8e0d359bf4edb84d5aa408cc3fa54a18c25ca74cd6e7
     steps:
-      - name: Checkout repository
+      - name: Check isolated checkout paths
+        run: |
+          test ! -e observer-source
+          test ! -L observer-source
+          test ! -e test-source
+          test ! -L test-source
+
+      - name: Checkout observer
         uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
         with:
+          path: observer-source
+          clean: false
           persist-credentials: false
 
-      - name: Setup Rust environment
-        uses: ./.github/actions/setup
-        with:
-          rust-version: stable
-          # Dedicated key: ubuntu-latest and sm-standard-4 share runner.os, so
-          # a shared key would mix VM and ARC pod target/ artifacts.
-          cache-shared-key: ci-e2e-distributed-hosted
-          cache-save-if: ${{ github.ref == 'refs/heads/main' }}
-          install-build-packaging-tools: 'false'
-
-      - name: Prepare isolated filesystems for pool movement
+      - name: Stage observer outside source
+        env:
+          WORKFLOW_SHA: ${{ github.sha }}
+        run: |
+          set -euo pipefail
+          test "$(git -C observer-source rev-parse HEAD)" = "$WORKFLOW_SHA"
+          mkdir -p "$RUNNER_TEMP/tls-private" "$RUNNER_TEMP/tls-evidence"
+          chmod 700 "$RUNNER_TEMP/tls-private"
+          cp observer-source/.github/scripts/observe_tls_resource.py "$RUNNER_TEMP/tls-private/observer.py"
+          python3 - <<'PY'
+          import hashlib, json, os
+          from pathlib import Path
+          workflow = Path('observer-source/.github/workflows/e2e-distributed.yml').read_bytes()
+          observer = Path('observer-source/.github/scripts/observe_tls_resource.py').read_bytes()
+          result = {'workflow_commit': os.environ['WORKFLOW_SHA'],
+                    'workflow_sha256': hashlib.sha256(workflow).hexdigest(),
+                    'observer_sha256': hashlib.sha256(observer).hexdigest()}
+          Path(os.environ['RUNNER_TEMP'], 'tls-evidence', 'workflow-identity.json').write_text(json.dumps(result, sort_keys=True) + '\n')
+          PY
+
+      - name: Checkout fixed test source
+        id: source
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          ref: fedc9666123076f2b985d75a9c290f6496ef8717
+          path: test-source
+          clean: false
+          persist-credentials: false
+
+      - name: Verify source commit and tree
+        id: identity
+        working-directory: test-source
+        run: |
+          set -euo pipefail
+          test "$(git rev-parse HEAD)" = fedc9666123076f2b985d75a9c290f6496ef8717
+          test "$(git rev-parse 'HEAD^{tree}')" = 1ea486f6d6a86ce488b83cab4a1b232188ab9708
+          test -z "$(git status --porcelain=v1 --untracked-files=all)"
+
+      # The repository's composite setup defaults its cache workspace to the
+      # checkout root. These are the same pinned tools, with the cache scoped
+      # explicitly to the independent source checkout.
+      - name: Install system dependencies
+        id: dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y build-essential pkg-config libssl-dev ripgrep
+
+      - name: Install protoc
+        id: protoc
+        uses: rustfs/setup-protoc@a3705324d8f9bf5b6c3573fb6cf8ae421db55dd6 # v3.0.1
+        with:
+          version: "35.1"
+          repo-token: ${{ github.token }}
+
+      - name: Install flatc
+        id: flatc
+        uses: Nugine/setup-flatc@698800de72a96bfb22cf60431dc21a2ff9a7e07b # v1
+        with:
+          version: "25.12.19"
+
+      - name: Install Rust toolchain
+        id: rust
+        uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
+        with:
+          toolchain: stable
+          components: rustfmt, clippy
+
+      - name: Install cargo-nextest
+        id: nextest
+        uses: taiki-e/install-action@96c7780c1d8a2b8723e12031def873a434d39d8d # nextest
+        with:
+          tool: cargo-nextest@0.9.138
+
+      - name: Restore Rust cache
+        id: setup
+        uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2
+        with:
+          workspaces: test-source -> target
+          shared-key: ci-e2e-distributed-hosted
+          cache-all-crates: false
+          cache-on-failure: true
+          save-if: false
+
+      # Match the original distributed lane; this case keeps its disks in the
+      # normal temporary directory, not in these otherwise empty pool mounts.
+      - name: Prepare isolated pool filesystems
+        id: mounts
         run: |
           set -euo pipefail
-          mount_base="${RUNNER_TEMP}/rustfs-e2e-pools"
-          mkdir -p "${mount_base}"
           roots=()
           for pool in 0 1 2 3; do
-            mountpoint="${mount_base}/pool-${pool}"
-            mkdir -p "${mountpoint}"
-            # Sized tmpfs reports a distinct st_dev and independent 1G
-            # statfs capacity. Requires a VM runner (ubuntu-latest).
-            if ! sudo mount -t tmpfs -o size=1G,nosuid,nodev,mode=1777 tmpfs "${mountpoint}"; then
-              echo "tmpfs mount failed on $(uname -a)" >&2
-              findmnt || true
-              grep Cap /proc/self/status || true
-              exit 1
-            fi
-            sudo chmod 1777 "${mountpoint}"
-            roots+=("${mountpoint}")
+            mountpoint="$RUNNER_TEMP/rustfs-e2e-pools/pool-$pool"
+            mkdir -p "$mountpoint"
+            sudo mount -t tmpfs -o size=1G,nosuid,nodev,mode=1777 tmpfs "$mountpoint"
+            sudo chmod 1777 "$mountpoint"
+            roots+=("$mountpoint")
           done
           printf -v joined_roots '%s:' "${roots[@]}"
-          echo "RUSTFS_E2E_POOL_ROOTS=${joined_roots%:}" >> "${GITHUB_ENV}"
-          findmnt --noheadings --output TARGET,SOURCE,FSTYPE,SIZE --target "${roots[0]}"
-          findmnt --noheadings --output TARGET,SOURCE,FSTYPE,SIZE --target "${roots[1]}"
-          findmnt --noheadings --output TARGET,SOURCE,FSTYPE,SIZE --target "${roots[2]}"
-          findmnt --noheadings --output TARGET,SOURCE,FSTYPE,SIZE --target "${roots[3]}"
+          echo "RUSTFS_E2E_POOL_ROOTS=${joined_roots%:}" >> "$GITHUB_ENV"
 
-      - name: Download pinned previous release
-        env:
-          SOURCE_DIR: ${{ runner.temp }}/rustfs-upgrade-source
+      - name: Build identified server
+        id: build
+        working-directory: test-source
         run: |
           set -euo pipefail
-          mkdir -p "$SOURCE_DIR"
-          archive="$SOURCE_DIR/$UPGRADE_SOURCE_ASSET"
-          curl --fail --location --retry 3 --output "$archive" \
-            "https://github.com/${GITHUB_REPOSITORY}/releases/download/${UPGRADE_SOURCE_VERSION}/${UPGRADE_SOURCE_ASSET}"
-          echo "$UPGRADE_SOURCE_SHA256  $archive" | sha256sum --check --strict
-          unzip -q "$archive" -d "$SOURCE_DIR"
-          chmod +x "$SOURCE_DIR/rustfs"
-          test -x "$SOURCE_DIR/rustfs"
-          echo "RUSTFS_UPGRADE_SOURCE_BINARY=$SOURCE_DIR/rustfs" >> "$GITHUB_ENV"
-
-      - name: Build rustfs binary
-        run: |
-          python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks
-
-      - name: Verify distributed e2e membership
-        env:
-          NEXTEST_LISTING: ${{ runner.temp }}/rustfs-e2e-distributed-list.json
-        run: |
-          cargo nextest list --profile e2e-distributed -p e2e_test --message-format json > "${NEXTEST_LISTING}"
-          python3 ./scripts/check_test_wiring.py --check-profile e2e-distributed "${NEXTEST_LISTING}"
-
-      - name: Run distributed 4-node e2e suite
-        env:
-          RUSTFS_E2E_LOG_DIR: ${{ runner.temp }}/rustfs-e2e-distributed-logs
-          FILTER: ${{ inputs.filter }}
-        run: |
-          set -euo pipefail
-          rm -f target/nextest/e2e-distributed/junit.xml
-          if [ -n "${FILTER}" ]; then
-            python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-distributed -p e2e_test -E "${FILTER}"
-          else
-            python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-distributed -p e2e_test --no-tests=fail
+          test "$(git rev-parse HEAD)" = fedc9666123076f2b985d75a9c290f6496ef8717
+          test -z "$(git status --porcelain=v1 --untracked-files=all)"
+          if ! python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks >"$RUNNER_TEMP/tls-private/build.log" 2>&1; then
+            echo "Server build failed; raw output withheld."
+            exit 1
           fi
-          test -s target/nextest/e2e-distributed/junit.xml
 
-      - name: Upload distributed e2e diagnostics
+      - name: Observe the single TLS test
+        id: observe
+        working-directory: test-source
+        env:
+          WORKFLOW_SHA: ${{ github.sha }}
+          RUN_ID: ${{ github.run_id }}
+          RUN_ATTEMPT: ${{ github.run_attempt }}
+        run: python3 "$RUNNER_TEMP/tls-private/observer.py"
+
+      - name: Record workflow stage outcomes
+        if: always()
+        env:
+          SOURCE_RESULT: ${{ steps.source.outcome }}
+          IDENTITY_RESULT: ${{ steps.identity.outcome }}
+          DEPENDENCIES_RESULT: ${{ steps.dependencies.outcome }}
+          PROTOC_RESULT: ${{ steps.protoc.outcome }}
+          FLATC_RESULT: ${{ steps.flatc.outcome }}
+          RUST_RESULT: ${{ steps.rust.outcome }}
+          NEXTEST_RESULT: ${{ steps.nextest.outcome }}
+          SETUP_RESULT: ${{ steps.setup.outcome }}
+          MOUNTS_RESULT: ${{ steps.mounts.outcome }}
+          BUILD_RESULT: ${{ steps.build.outcome }}
+          OBSERVE_RESULT: ${{ steps.observe.outcome }}
+        run: |
+          mkdir -p "$RUNNER_TEMP/tls-evidence"
+          python3 - <<'PY'
+          import json, os
+          from pathlib import Path
+          allowed = {'success', 'failure', 'cancelled', 'skipped'}
+          names = ('SOURCE', 'IDENTITY', 'DEPENDENCIES', 'PROTOC', 'FLATC', 'RUST', 'NEXTEST', 'SETUP', 'MOUNTS', 'BUILD', 'OBSERVE')
+          outcomes = {name.lower(): os.environ.get(name + '_RESULT') for name in names}
+          outcomes = {name: value if value in allowed else 'missing' for name, value in outcomes.items()}
+          Path(os.environ['RUNNER_TEMP'], 'tls-evidence', 'stages.json').write_text(json.dumps(outcomes, sort_keys=True) + '\n')
+          PY
+
+      - name: Upload sanitized numerical evidence
         if: always()
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
         with:
-          name: e2e-distributed-${{ github.run_number }}
-          path: |
-            target/nextest/e2e-distributed/junit.xml
-            ${{ runner.temp }}/rustfs-e2e-distributed-list.json
-            ${{ runner.temp }}/rustfs-e2e-distributed-logs/
+          name: tls-resource-observation-${{ github.run_id }}-${{ github.run_attempt }}
+          path: ${{ runner.temp }}/tls-evidence/
           retention-days: 7
-          if-no-files-found: warn
+          if-no-files-found: error
 
       - name: Unmount isolated pool filesystems
         if: always()
         run: |
           set -euo pipefail
-          mount_base="${RUNNER_TEMP}/rustfs-e2e-pools"
           for pool in 0 1 2 3; do
-            mountpoint="${mount_base}/pool-${pool}"
-            if mountpoint --quiet "${mountpoint}"; then
-              sudo umount "${mountpoint}"
+            mountpoint="$RUNNER_TEMP/rustfs-e2e-pools/pool-$pool"
+            if mountpoint --quiet "$mountpoint"; then
+              sudo umount "$mountpoint"
             fi
           done
-
-  alert-on-failure:
-    name: Alert on scheduled failure
-    needs: [distributed]
-    if: github.repository == 'rustfs/rustfs' && (always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure'))
-    runs-on: ubuntu-latest
-    timeout-minutes: 10
-    permissions:
-      contents: read
-      issues: write
-    steps:
-      - uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
-        with:
-          persist-credentials: false
-      - name: Open or update failure-tracking issue
-        uses: ./.github/actions/schedule-failure-issue
-        with:
-          github-token: ${{ secrets.BACKLOG_ISSUE_TOKEN }}

crates/e2e_test/src/distributed/internode_tls_test.rs

@@ -3,13 +3,14 @@
 
 use super::harness::{DistCluster, DistLayout, TestResult, unique_bucket};
 use crate::tls_gen;
+use aws_sdk_s3::operation::RequestId;
 use aws_sdk_s3::{Client, config::retry::RetryConfig, primitives::ByteStream};
 use rand::{Rng, SeedableRng, rngs::StdRng};
 use sha2::{Digest, Sha256};
 use std::io::{Read, Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
-use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::sync::{Arc, Mutex, TryLockError};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use tokio::task::JoinSet;
 
 const MIB: u64 = 1024 * 1024;
@@ -28,6 +29,285 @@ struct ObjectFixture {
     reads: Vec<ExpectedRead>,
 }
 
+#[derive(Clone)]
+struct ConcurrentReadStream {
+    stage: &'static str,
+    started_ms: Option<u64>,
+    headers_ms: Option<u64>,
+    body_started_ms: Option<u64>,
+    completed_ms: Option<u64>,
+    received: u64,
+    last_progress_ms: Option<u64>,
+    request_id: Option<uuid::Uuid>,
+    terminal: Option<&'static str>,
+    terminal_ms: Option<u64>,
+}
+
+impl Default for ConcurrentReadStream {
+    fn default() -> Self {
+        Self {
+            stage: "queued",
+            started_ms: None,
+            headers_ms: None,
+            body_started_ms: None,
+            completed_ms: None,
+            received: 0,
+            last_progress_ms: None,
+            request_id: None,
+            terminal: None,
+            terminal_ms: None,
+        }
+    }
+}
+
+struct ConcurrentReadProgress {
+    started: Instant,
+    started_clock: ClockSample,
+    expected: u64,
+    streams: Vec<Mutex<ConcurrentReadStream>>,
+}
+
+#[derive(Clone, Copy)]
+struct ClockSample {
+    unix_ms: Option<u64>,
+    before_ms: Option<u64>,
+    after_ms: Option<u64>,
+}
+
+impl ClockSample {
+    fn capture(started: Instant) -> Self {
+        let before_ms = elapsed_ms(started);
+        Self::capture_after(started, before_ms)
+    }
+
+    fn capture_after(started: Instant, before_ms: Option<u64>) -> Self {
+        let unix_ms = unix_ms_at(SystemTime::now());
+        let after_ms = elapsed_ms(started);
+        Self {
+            unix_ms,
+            before_ms,
+            after_ms,
+        }
+    }
+
+    fn json(self) -> serde_json::Value {
+        serde_json::json!({
+            "unix_ms": self.unix_ms,
+            "monotonic_before_ms": self.before_ms,
+            "monotonic_after_ms": self.after_ms,
+        })
+    }
+}
+
+impl ConcurrentReadProgress {
+    fn new(expected: u64, count: usize) -> Self {
+        let started = Instant::now();
+        Self {
+            started,
+            started_clock: ClockSample::capture(started),
+            expected,
+            streams: (0..count).map(|_| Mutex::new(ConcurrentReadStream::default())).collect(),
+        }
+    }
+
+    fn elapsed_ms(&self) -> Option<u64> {
+        elapsed_ms(self.started)
+    }
+
+    fn start(&self, node: usize) {
+        let mut stream = self.streams[node].lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        stream.started_ms = self.elapsed_ms();
+        stream.stage = "waiting_headers";
+    }
+
+    fn response(&self, node: usize, request_id: Option<&str>, has_headers: bool) {
+        let mut stream = self.streams[node].lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        if let Some(request_id) = request_id.and_then(|id| uuid::Uuid::parse_str(id).ok()) {
+            stream.request_id = Some(request_id);
+        }
+        if has_headers {
+            stream.headers_ms = self.elapsed_ms();
+            stream.stage = "headers_received";
+        }
+    }
+
+    fn body(&self, node: usize) {
+        let mut stream = self.streams[node].lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        stream.body_started_ms = self.elapsed_ms();
+        stream.stage = "reading_body";
+    }
+
+    fn progress(&self, node: usize, received: u64) {
+        let mut stream = self.streams[node].lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        stream.last_progress_ms = self.elapsed_ms();
+        stream.received = received;
+    }
+
+    fn complete(&self, node: usize) {
+        let mut stream = self.streams[node].lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        stream.completed_ms = self.elapsed_ms();
+        stream.stage = "complete";
+    }
+
+    fn terminal(&self, node: usize, terminal: &'static str) {
+        let mut stream = self.streams[node].lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        stream.terminal = Some(terminal);
+        stream.terminal_ms = self.elapsed_ms();
+    }
+
+    fn failure_snapshot(&self, failed_node: Option<usize>, kind: &'static str) -> serde_json::Value {
+        let streams: Vec<_> = self
+            .streams
+            .iter()
+            .enumerate()
+            .map(|(node, stream)| {
+                // Avoid waiting on another task while retaining the first error.
+                let mut snapshot = None;
+                let mut unavailable = "busy";
+                for _ in 0..8 {
+                    let before_ms = self.elapsed_ms();
+                    match stream.try_lock() {
+                        Ok(guard) => {
+                            let stream = guard.clone();
+                            drop(guard);
+                            snapshot = Some((stream, ClockSample::capture_after(self.started, before_ms)));
+                            break;
+                        }
+                        Err(TryLockError::WouldBlock) => std::hint::spin_loop(),
+                        Err(TryLockError::Poisoned(_)) => {
+                            unavailable = "poisoned";
+                            break;
+                        }
+                    }
+                }
+                let Some((stream, sampled_clock)) = snapshot else {
+                    return serde_json::json!({
+                        "node": node,
+                        "snapshot": unavailable,
+                        "sampled_clock": ClockSample::capture(self.started).json(),
+                        "last_stage": "unknown",
+                        "terminal": "unknown",
+                        "request_id": null,
+                        "received_bytes": null,
+                        "last_progress_ms": null,
+                    });
+                };
+                serde_json::json!({
+                    "node": node,
+                    "snapshot": "available",
+                    "sampled_clock": sampled_clock.json(),
+                    "last_stage": stream.stage,
+                    "request_id": stream.request_id.map(|id| id.to_string()),
+                    "started_ms": stream.started_ms,
+                    "headers_ms": stream.headers_ms,
+                    "body_started_ms": stream.body_started_ms,
+                    "completed_ms": stream.completed_ms,
+                    "received_bytes": stream.received,
+                    "last_progress_ms": stream.last_progress_ms,
+                    "terminal": stream.terminal.unwrap_or("unknown"),
+                    "terminal_ms": stream.terminal_ms,
+                })
+            })
+            .collect();
+        serde_json::json!({
+            "kind": kind,
+            "failed_node": failed_node,
+            "elapsed_ms": self.elapsed_ms(),
+            "stage_start_clock_sample": self.started_clock.json(),
+            "clock_scope": "client_unsynchronized",
+            "expected_bytes": self.expected,
+            "streams": streams,
+        })
+    }
+
+    fn report_failure(&self, failed_node: Option<usize>, kind: &'static str) {
+        eprintln!("TLS acceptance: concurrent GET failure: {}", self.failure_snapshot(failed_node, kind));
+    }
+}
+
+fn elapsed_ms(started: Instant) -> Option<u64> {
+    u64::try_from(started.elapsed().as_millis()).ok()
+}
+
+fn unix_ms_at(time: SystemTime) -> Option<u64> {
+    time.duration_since(UNIX_EPOCH)
+        .ok()
+        .and_then(|duration| u64::try_from(duration.as_millis()).ok())
+}
+
+fn verify_concurrent_read_failure_snapshot_evidence() {
+    let progress = ConcurrentReadProgress::new(100, 4);
+    progress.start(0);
+    progress.response(0, Some("LEAK-invalid-request-id"), true);
+    progress.body(0);
+    progress.progress(0, 100);
+    progress.complete(0);
+    progress.terminal(0, "success");
+
+    let valid_id = "b8172a15-a428-4b42-9cef-1f4a7b96874e";
+    progress.start(1);
+    progress.response(1, Some(valid_id), true);
+    progress.body(1);
+    progress.progress(1, 41);
+    progress.terminal(1, "failed");
+
+    progress.start(3);
+    let _busy = progress.streams[2].lock().expect("hold one stream during the snapshot");
+    let snapshot = progress.failure_snapshot(Some(1), "read_failed");
+    assert_eq!(snapshot["clock_scope"], "client_unsynchronized");
+    assert_eq!(snapshot["failed_node"], 1);
+    let start_clock = &snapshot["stage_start_clock_sample"];
+    assert!(
+        start_clock["monotonic_before_ms"].as_u64().expect("start sample lower bound")
+            <= start_clock["monotonic_after_ms"].as_u64().expect("start sample upper bound")
+    );
+    assert!(!snapshot.to_string().contains("LEAK-invalid-request-id"));
+
+    let streams = snapshot["streams"].as_array().expect("per-stream snapshots");
+    assert_eq!(streams.len(), 4);
+    for (stream, terminal, expected_bytes) in [(&streams[0], "success", 100), (&streams[1], "failed", 41)] {
+        assert_eq!(stream["snapshot"], "available");
+        assert_eq!(stream["terminal"], terminal);
+        assert_eq!(stream["received_bytes"], expected_bytes);
+        let times = [
+            "started_ms",
+            "headers_ms",
+            "body_started_ms",
+            "last_progress_ms",
+            "terminal_ms",
+        ]
+        .map(|field| stream[field].as_u64().expect("observed stream time"));
+        assert!(times.windows(2).all(|pair| pair[0] <= pair[1]));
+        let clock = &stream["sampled_clock"];
+        let before = clock["monotonic_before_ms"].as_u64().expect("sample lower bound");
+        let after = clock["monotonic_after_ms"].as_u64().expect("sample upper bound");
+        assert!(times[4] <= after && before <= after);
+    }
+    assert_eq!(streams[0]["last_stage"], "complete");
+    assert!(streams[0]["request_id"].is_null());
+    assert!(
+        streams[0]["completed_ms"].as_u64().expect("completion time")
+            <= streams[0]["terminal_ms"].as_u64().expect("terminal time")
+    );
+    assert_eq!(streams[1]["last_stage"], "reading_body");
+    assert!(streams[1]["completed_ms"].is_null());
+    assert_eq!(streams[1]["request_id"], valid_id);
+    assert_eq!(streams[2]["snapshot"], "busy");
+    assert_eq!(streams[2]["last_stage"], "unknown");
+    assert_eq!(streams[2]["terminal"], "unknown");
+    assert!(streams[2]["received_bytes"].is_null());
+    assert!(streams[2]["sampled_clock"]["monotonic_before_ms"].is_number());
+    assert_eq!(streams[3]["last_stage"], "waiting_headers");
+    assert_eq!(streams[3]["terminal"], "unknown");
+    assert!(streams[3]["request_id"].is_null());
+    assert!(streams[3]["last_progress_ms"].is_null());
+
+    let before_epoch = UNIX_EPOCH
+        .checked_sub(Duration::from_millis(1))
+        .expect("represent pre-epoch time");
+    assert!(unix_ms_at(before_epoch).is_none());
+}
+
 fn fixture(directory: &Path, length: u64) -> TestResult<ObjectFixture> {
     let key = format!("object-{length}");
     let path = directory.join(&key);
@@ -85,16 +365,28 @@ async fn verified_get(
     object: &ObjectFixture,
     expected: &ExpectedRead,
     concurrent: bool,
+    progress: Option<&ConcurrentReadProgress>,
 ) -> TestResult<serde_json::Value> {
     let started = Instant::now();
     let mut received = 0u64;
     let mut stage = "response headers";
+    if let Some(progress) = progress {
+        progress.start(node);
+    }
     tokio::time::timeout(READ_TIMEOUT, async {
         let mut request = client.get_object().bucket(bucket).key(&object.key);
         if let Some((start, end)) = expected.range {
             request = request.range(format!("bytes={start}-{end}"));
         }
-        let mut output = request.send().await?;
+        let response = request.send().await;
+        if let Some(progress) = progress {
+            let has_headers = match &response {
+                Ok(_) => true,
+                Err(error) => error.raw_response().is_some(),
+            };
+            progress.response(node, response.request_id(), has_headers);
+        }
+        let mut output = response?;
         assert_eq!(output.content_length(), Some(i64::try_from(expected.length)?));
         if let Some((start, end)) = expected.range {
             assert_eq!(output.content_range(), Some(format!("bytes {start}-{end}/{}", object.length).as_str()));
@@ -102,21 +394,32 @@ async fn verified_get(
             assert_eq!(output.content_range(), None);
         }
         stage = "response body";
+        if let Some(progress) = progress {
+            progress.body(node);
+        }
         let mut digest = Sha256::new();
         while let Some(chunk) = output.body.next().await {
-            let chunk = chunk.inspect_err(|error| {
+            let chunk = chunk.inspect_err(|_| {
                 eprintln!(
-                    "TLS acceptance: GET body failed: node={node}, key={}, range={:?}, concurrent={concurrent}, received={received}/{}, elapsed={:?}, error={error}",
+                    "TLS acceptance: GET body failed: node={node}, key={}, range={:?}, concurrent={concurrent}, received={received}/{}, elapsed={:?}",
                     object.key, expected.range, expected.length, started.elapsed()
                 );
             })?;
             received += u64::try_from(chunk.len())?;
+            if !chunk.is_empty()
+                && let Some(progress) = progress
+            {
+                progress.progress(node, received);
+            }
             assert!(received <= expected.length, "node {node}: GET exceeded the requested length");
             digest.update(&chunk);
         }
         let actual_sha256 = hex_simd::encode_to_string(digest.finalize(), hex_simd::AsciiCase::Lower);
         assert_eq!(received, expected.length, "node {node}: truncated GET {}", object.key);
         assert_eq!(actual_sha256, expected.sha256, "node {node}: corrupt GET {}", object.key);
+        if let Some(progress) = progress {
+            progress.complete(node);
+        }
         eprintln!(
             "TLS acceptance: GET node={node}, key={}, range={:?}, concurrent={concurrent}, bytes={received}, elapsed={:?}",
             object.key, expected.range, started.elapsed()
@@ -143,6 +446,7 @@ async fn verified_get(
 /// tiny objects and count that as a large distributed-read pass.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn four_node_tls_large_and_range_gets_preserve_bytes() -> TestResult {
+    verify_concurrent_read_failure_snapshot_evidence();
     let mode = std::env::var("RUSTFS_E2E_INTERNODE_HTTP_MODE").unwrap_or_else(|_| "http1".into());
     let protocol = match mode.as_str() {
         "http1" => "http1",
@@ -238,7 +542,7 @@ async fn four_node_tls_large_and_range_gets_preserve_bytes() -> TestResult {
         if length == 32 * 1024 {
             let alternate = dist.client_with_credentials(0, &dist.cluster.access_key, &dist.cluster.secret_key)?;
             let alternate = Client::from_conf(alternate.config().to_builder().retry_config(RetryConfig::disabled()).build());
-            results.push(verified_get(&alternate, 0, &bucket, &object, &object.reads[0], false).await?);
+            results.push(verified_get(&alternate, 0, &bucket, &object, &object.reads[0], false, None).await?);
             let invalid = dist.client_with_credentials(0, "invalid-tls-user", "invalid-tls-secret-password")?;
             let invalid = Client::from_conf(invalid.config().to_builder().retry_config(RetryConfig::disabled()).build());
             let denied = invalid
@@ -250,21 +554,45 @@ async fn four_node_tls_large_and_range_gets_preserve_bytes() -> TestResult {
         }
         for expected in &object.reads {
             for (node, client) in clients.iter().enumerate() {
-                results.push(verified_get(client, node, &bucket, &object, expected, false).await?);
+                results.push(verified_get(client, node, &bucket, &object, expected, false, None).await?);
             }
         }
         // Each reader owns a Tokio task and its cooperative budget. Polling
         // every body and synchronous digest in one parent task can let an
         // always-ready reader starve its siblings and their deadlines.
         let mut readers = JoinSet::new();
+        let progress = Arc::new(ConcurrentReadProgress::new(object.reads[0].length, clients.len()));
+        let mut task_nodes = Vec::with_capacity(clients.len());
         for (node, client) in clients.iter().enumerate() {
             let client = client.clone();
             let bucket = bucket.clone();
             let object = Arc::clone(&object);
-            readers.spawn(async move { verified_get(&client, node, &bucket, &object, &object.reads[0], true).await });
+            let progress = Arc::clone(&progress);
+            let handle = readers.spawn(async move {
+                let result = verified_get(&client, node, &bucket, &object, &object.reads[0], true, Some(&progress)).await;
+                progress.terminal(node, if result.is_ok() { "success" } else { "failed" });
+                (node, result)
+            });
+            task_nodes.push((handle.id(), node));
         }
         while let Some(result) = readers.join_next().await {
-            results.push(result??);
+            match result {
+                Ok((_, Ok(result))) => results.push(result),
+                Ok((node, Err(error))) => {
+                    readers.abort_all();
+                    progress.report_failure(Some(node), "read_failed");
+                    return Err(error);
+                }
+                Err(error) => {
+                    let node = task_nodes.iter().find(|(id, _)| *id == error.id()).map(|(_, node)| *node);
+                    readers.abort_all();
+                    if let Some(node) = node {
+                        progress.terminal(node, if error.is_panic() { "panicked" } else { "cancelled" });
+                    }
+                    progress.report_failure(node, "task_failed");
+                    return Err(error.into());
+                }
+            }
         }
         tokio::fs::remove_file(&object.path).await?;
     }

crates/scanner/src/heal_completion.rs

@@ -183,17 +183,7 @@ impl Settlement {
             "insufficient_scan"
         } else {
             match (candidate.identity.as_ref(), completion.object.as_ref()) {
-                (Some(expected), Some(actual)) if expected != &actual.identity => "identity_mismatch",
-                (Some(expected), Some(_))
-                    if expected.bucket_incarnation_id.is_some_and(|id| !id.is_nil())
-                        && expected.pool_index.is_some()
-                        && expected.set_index.is_some()
-                        && expected.version_id.as_ref().is_some_and(|version| !version.is_empty()) =>
-                {
-                    // Even an exact version and VerifiedHealthy do not bind the
-                    // candidate's content and leader epoch with today's contract.
-                    "content_and_leader_unknown"
-                }
+                (Some(expected), Some(actual)) => Self::identity_scope(expected, &actual.identity),
                 _ => "identity_unknown",
             }
         };
@@ -209,6 +199,46 @@ impl Settlement {
         }
     }
 
+    fn identity_scope(expected: &HealObjectIdentity, actual: &HealObjectIdentity) -> &'static str {
+        let expected_version = expected.version_id.as_deref().filter(|version| !version.is_empty());
+        let actual_version = actual.version_id.as_deref().filter(|version| !version.is_empty());
+        let expected_incarnation = expected.bucket_incarnation_id.filter(|id| !id.is_nil());
+        let actual_incarnation = actual.bucket_incarnation_id.filter(|id| !id.is_nil());
+        if expected.kind != actual.kind
+            || expected.bucket != actual.bucket
+            || expected.object != actual.object
+            || expected_version
+                .zip(actual_version)
+                .is_some_and(|(left, right)| left != right)
+            || expected_incarnation
+                .zip(actual_incarnation)
+                .is_some_and(|(left, right)| left != right)
+            || expected
+                .pool_index
+                .zip(actual.pool_index)
+                .is_some_and(|(left, right)| left != right)
+            || expected
+                .set_index
+                .zip(actual.set_index)
+                .is_some_and(|(left, right)| left != right)
+        {
+            "identity_mismatch"
+        } else if expected_version.is_none()
+            || actual_version.is_none()
+            || expected_incarnation.is_none()
+            || actual_incarnation.is_none()
+            || expected.pool_index.is_none()
+            || actual.pool_index.is_none()
+            || expected.set_index.is_none()
+            || actual.set_index.is_none()
+        {
+            "identity_unknown"
+        } else {
+            // Matching identities still do not bind content and leader epoch.
+            "content_and_leader_unknown"
+        }
+    }
+
     fn record(&self) {
         counter!("rustfs_scanner_heal_candidate_terminal_total",
             "disposition" => self.disposition, "scope" => self.scope, "scan_mode" => self.mode.as_str())

crates/scanner/src/heal_completion/tests.rs

@@ -98,6 +98,91 @@ fn replacement_bucket_version_or_location_cannot_match_an_old_result() {
     }
 }
 
+#[test]
+fn missing_identity_fields_are_unknown_on_either_side() {
+    type IdentityChange = fn(&mut HealObjectIdentity);
+    let changes: [(&str, IdentityChange); 6] = [
+        ("version", |identity| identity.version_id = None),
+        ("empty version", |identity| identity.version_id = Some(String::new())),
+        ("incarnation", |identity| identity.bucket_incarnation_id = None),
+        ("nil incarnation", |identity| identity.bucket_incarnation_id = Some(Uuid::nil())),
+        ("pool", |identity| identity.pool_index = None),
+        ("set", |identity| identity.set_index = None),
+    ];
+    for (label, change) in changes {
+        for candidate_missing in [true, false] {
+            let mut observed = candidate();
+            let mut terminal = (*completion("partial", HealObjectDisposition::VerifiedHealthy)).clone();
+            if candidate_missing {
+                change(observed.identity.as_mut().expect("candidate identity"));
+            } else {
+                change(&mut terminal.object.as_mut().expect("receipt").identity);
+            }
+            let mut recent = Recent {
+                completion: Arc::new(terminal),
+                counted: false,
+            };
+            assert_eq!(
+                Settlement::new(observed, &mut recent).scope,
+                "identity_unknown",
+                "{label}, candidate missing: {candidate_missing}"
+            );
+        }
+    }
+}
+
+#[test]
+fn known_identity_conflicts_are_not_hidden_by_other_missing_fields() {
+    for change in 0..3 {
+        let mut observed = candidate();
+        let mut terminal = (*completion("conflict", HealObjectDisposition::VerifiedHealthy)).clone();
+        let expected = observed.identity.as_mut().expect("candidate identity");
+        let actual = &mut terminal.object.as_mut().expect("receipt").identity;
+        match change {
+            0 => {
+                expected.version_id = None;
+                actual.bucket = "replacement".into();
+            }
+            1 => {
+                expected.pool_index = None;
+                actual.version_id = Some(Uuid::from_u128(12).to_string());
+            }
+            _ => {
+                expected.bucket_incarnation_id = None;
+                actual.set_index = Some(2);
+            }
+        }
+        let mut recent = Recent {
+            completion: Arc::new(terminal),
+            counted: false,
+        };
+        assert_eq!(Settlement::new(observed, &mut recent).scope, "identity_mismatch", "case {change}");
+    }
+}
+
+#[test]
+fn scanner_candidate_without_location_records_identity_unknown() {
+    let mut scanner_request = request();
+    scanner_request.expected_bucket_incarnation_id = None;
+    scanner_request.pool_index = None;
+    scanner_request.set_index = None;
+    let mut observations = Observations::default();
+    assert!(
+        observations
+            .admit(
+                "scanner".into(),
+                Candidate::from_request(&scanner_request).expect("bounded scanner candidate"),
+            )
+            .is_none()
+    );
+    let settlement = observations
+        .complete(completion("scanner", HealObjectDisposition::VerifiedHealthy), Instant::now())
+        .expect("terminal candidate observed");
+    assert_eq!(settlement.scope, "identity_unknown");
+    assert_eq!(observations.unknown_counts.get("identity_unknown"), Some(&1));
+    assert_eq!(observations.unknown_counts.get("identity_mismatch"), None);
+}
+
 #[test]
 fn unversioned_overwrite_and_unknown_leader_never_certify_payload() {
     let mut observed = candidate();