ci: isolate TLS diagnostic report environment

test: create private TLS observation fixture atomically
test: capture bounded TLS read observations
2026-11-02 17:13:37 +00:00 · 2026-10-28 11:32:05 +08:00 · 2026-10-28 10:07:29 +08:00 · 2026-10-28 09:25:01 +08:00 · 2026-10-28 05:36:17 +08:00 · 2026-10-28 03:00:56 +08:00
12 changed files with 2510 additions and 211 deletions
@@ -1,219 +1,215 @@
-# 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
+name: TLS GET observation diagnostic

 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' }}
+  group: tls-get-observation-${{ github.ref }}
+  cancel-in-progress: false

 jobs:
-  distributed:
-    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").
+  tls-get:
    runs-on: ubuntu-latest
    timeout-minutes: 180
    env:
      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+      CARGO_TERM_COLOR: never
+      CARGO_INCREMENTAL: "0"
+      CARGO_BUILD_JOBS: "2"
      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
+      FILTER: "test(=distributed::internode_tls_test::four_node_tls_large_and_range_gets_preserve_bytes)"
    steps:
-      - name: Checkout repository
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+      - name: Checkout
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a
        with:
+          fetch-depth: 3
          persist-credentials: false
+          clean: false

-      - name: Setup Rust environment
-        uses: ./.github/actions/setup
+      - name: Verify source snapshot
+        shell: bash
+        run: |
+          set -euo pipefail
+          workflow_sha=$(git rev-parse HEAD)
+          source_sha=$(git rev-parse HEAD^)
+          base_sha=$(git rev-parse HEAD^^)
+          source_tree=$(git rev-parse "$source_sha^{tree}")
+          source_diff=$(git diff --full-index --binary "$base_sha" "$source_sha" | sha256sum | cut -d' ' -f1)
+          workflow_files=$(git diff --name-only "$source_sha" "$workflow_sha")
+          printf 'head=%s source=%s base=%s tree=%s diff=%s\n' "$workflow_sha" "$source_sha" "$base_sha" "$source_tree" "$source_diff"
+          [[ "$workflow_sha" == "$GITHUB_SHA" ]]
+          [[ "$source_sha" == '0339b4f8815b4b6e94a1b6c0d6c29b4700d1eb2f' ]]
+          [[ "$base_sha" == '7d3f077072e43ff1ae1844a14c09dec623107f69' ]]
+          [[ "$source_tree" == '9a92610ac6e9c054c4a567a896784ddae5c6034c' ]]
+          [[ "$source_diff" == '8b4da474b948371493945bf05545f768b026870be340d410b89c4e4a605412f4' ]]
+          [[ "$workflow_files" == '.github/workflows/e2e-distributed.yml' ]]
+          [[ -z "$(git status --porcelain)" ]]
+          test ! -e target/nextest/e2e-distributed/junit.xml
+          test ! -L target/nextest/e2e-distributed/junit.xml
+
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y build-essential pkg-config libssl-dev ripgrep
+
+      - name: Install protoc
+        uses: rustfs/setup-protoc@a3705324d8f9bf5b6c3573fb6cf8ae421db55dd6
        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'
+          version: '35.1'
+          repo-token: ${{ github.token }}

-      - name: Prepare isolated filesystems for pool movement
+      - name: Install flatc
+        uses: Nugine/setup-flatc@698800de72a96bfb22cf60431dc21a2ff9a7e07b
+        with:
+          version: '25.12.19'
+
+      - name: Install Rust
+        uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8
+        with:
+          toolchain: stable
+
+      - name: Install nextest
+        uses: taiki-e/install-action@96c7780c1d8a2b8723e12031def873a434d39d8d
+        with:
+          tool: cargo-nextest@0.9.138
+
+      - name: Build identified RustFS binary
+        run: python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks
+
+      - name: Verify profile and exact selection
+        id: selection
+        shell: bash
+        env:
+          PROFILE_LIST: ${{ runner.temp }}/tls-get-distributed-list.json
+          SELECTED_LIST: ${{ runner.temp }}/tls-get-selected-list.json
        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}")
-          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]}"
+          test ! -e "$PROFILE_LIST"
+          test ! -L "$PROFILE_LIST"
+          test ! -e "$SELECTED_LIST"
+          test ! -L "$SELECTED_LIST"
+          echo 'fresh=true' >> "$GITHUB_OUTPUT"
+          cargo nextest list --profile e2e-distributed -p e2e_test --message-format json > "$PROFILE_LIST"
+          python3 scripts/check_test_wiring.py --check-profile e2e-distributed "$PROFILE_LIST"
+          cargo nextest list --profile e2e-distributed -p e2e_test -E "$FILTER" --message-format json > "$SELECTED_LIST"
+          python3 - "$SELECTED_LIST" <<'PY'
+          import json
+          import sys

-      - name: Download pinned previous release
+          with open(sys.argv[1], encoding="utf-8") as listing:
+              data = json.load(listing)
+          selected = [
+              (suite_id, name, case)
+              for suite_id, suite in data["rust-suites"].items()
+              for name, case in suite["testcases"].items()
+              if case.get("filter-match", {}).get("status") == "matches"
+          ]
+          expected = "distributed::internode_tls_test::four_node_tls_large_and_range_gets_preserve_bytes"
+          if len(selected) != 1 or selected[0][:2] != ("e2e_test", expected):
+              raise SystemExit("expected exactly the named TLS GET test")
+          if selected[0][2].get("kind") != "test" or selected[0][2].get("ignored") is not False:
+              raise SystemExit("the selected TLS GET test must be runnable")
+          print("Selected exactly one TLS GET test")
+          PY
+
+      # This case starts scanner and heal; normal background MRF may run.
+      # It does not select the rejected coordinator RPC or MRF release tests.
+      - name: Run single TLS GET diagnostic
+        id: diagnostic
+        shell: bash
        env:
-          SOURCE_DIR: ${{ runner.temp }}/rustfs-upgrade-source
+          TMPDIR: ${{ runner.temp }}/tls-get-fixtures
+          RUSTFS_E2E_LOG_DIR: ${{ runner.temp }}/tls-get-node-logs
+          RUSTFS_E2E_TLS_GET_OBSERVATION_DIR: ${{ runner.temp }}/tls-get-observation
        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"
+          unset RUSTFS_INTERNODE_ACCEPTANCE_REPORT
+          test ! -e target/nextest/e2e-distributed/junit.xml
+          test ! -L target/nextest/e2e-distributed/junit.xml
+          test ! -e "${RUNNER_TEMP}/tls-get-nextest.log"
+          test ! -L "${RUNNER_TEMP}/tls-get-nextest.log"
+          umask 077
+          mkdir -m 0700 "$TMPDIR"
+          mkdir -m 0700 "$RUSTFS_E2E_LOG_DIR"
+          mkdir -m 0700 "$RUSTFS_E2E_TLS_GET_OBSERVATION_DIR"
+          TMPDIR=$(realpath -e -- "$TMPDIR")
+          RUSTFS_E2E_LOG_DIR=$(realpath -e -- "$RUSTFS_E2E_LOG_DIR")
+          RUSTFS_E2E_TLS_GET_OBSERVATION_DIR=$(realpath -e -- "$RUSTFS_E2E_TLS_GET_OBSERVATION_DIR")
+          export TMPDIR RUSTFS_E2E_LOG_DIR RUSTFS_E2E_TLS_GET_OBSERVATION_DIR
+          python3 - "$TMPDIR" "$RUSTFS_E2E_LOG_DIR" "$RUSTFS_E2E_TLS_GET_OBSERVATION_DIR" <<'PY'
+          import os
+          from pathlib import Path
+          import stat
+          import sys

-      - 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
-          fi
+          roots = [Path(value).resolve(strict=True) for value in sys.argv[1:]]
+          for root in roots:
+              metadata = root.stat()
+              if not stat.S_ISDIR(metadata.st_mode) or metadata.st_uid != os.getuid() or stat.S_IMODE(metadata.st_mode) != 0o700:
+                  raise SystemExit("diagnostic directories must be private and owned by the runner")
+          for index, root in enumerate(roots):
+              for other in roots[index + 1:]:
+                  if root == other or root in other.parents or other in root.parents:
+                      raise SystemExit("fixture, log, and observation directories must be independent")
+          print("Fixture, log, and observation directories are private and independent")
+          PY
+          echo 'ready=true' >> "$GITHUB_OUTPUT"
+          python3 scripts/e2e_binary.py run --features e2e-test-hooks -- \
+            cargo nextest run --profile e2e-distributed -p e2e_test -E "$FILTER" \
+            --no-tests=fail --success-output=final --failure-output=final \
+            2>&1 | tee "${RUNNER_TEMP}/tls-get-nextest.log"
          test -s target/nextest/e2e-distributed/junit.xml
+          python3 - target/nextest/e2e-distributed/junit.xml <<'PY'
+          import sys
+          import xml.etree.ElementTree as ET

-      - name: Upload distributed e2e diagnostics
-        if: always()
-        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+          root = ET.parse(sys.argv[1]).getroot()
+          suites = root.findall("testsuite")
+          cases = root.findall(".//testcase")
+          expected = "distributed::internode_tls_test::four_node_tls_large_and_range_gets_preserve_bytes"
+          if root.tag != "testsuites" or any(root.get(key) != value for key, value in (
+              ("tests", "1"), ("failures", "0"), ("errors", "0")
+          )):
+              raise SystemExit("JUnit totals do not prove one successful test")
+          if len(suites) != 1 or any(suites[0].get(key) != value for key, value in (
+              ("name", "e2e_test"), ("tests", "1"), ("failures", "0"), ("errors", "0"), ("disabled", "0")
+          )):
+              raise SystemExit("JUnit suite does not prove one successful test")
+          if len(cases) != 1 or cases[0].get("name") != expected or cases[0].get("classname") != "e2e_test":
+              raise SystemExit("JUnit did not report the exact TLS GET test")
+          bad = {"failure", "error", "skipped", "flakyFailure", "rerunFailure", "rerunError"}
+          if any(element.tag.rsplit("}", 1)[-1] in bad for element in root.iter()):
+              raise SystemExit("JUnit contains a failed, skipped, or retried test")
+          print("JUnit confirms the exact test passed")
+          PY
+
+      - name: Preserve selection listings
+        if: always() && steps.selection.outputs.fresh == 'true'
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a
        with:
-          name: e2e-distributed-${{ github.run_number }}
+          name: tls-get-selection-${{ github.run_id }}-${{ github.run_attempt }}
          path: |
-            target/nextest/e2e-distributed/junit.xml
-            ${{ runner.temp }}/rustfs-e2e-distributed-list.json
-            ${{ runner.temp }}/rustfs-e2e-distributed-logs/
+            ${{ runner.temp }}/tls-get-distributed-list.json
+            ${{ runner.temp }}/tls-get-selected-list.json
          retention-days: 7
          if-no-files-found: warn

-      - 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}"
-            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
+      - name: Preserve diagnostic logs
+        if: always() && steps.diagnostic.outputs.ready == 'true'
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a
        with:
-          persist-credentials: false
-      - name: Open or update failure-tracking issue
-        uses: ./.github/actions/schedule-failure-issue
-        with:
-          github-token: ${{ secrets.BACKLOG_ISSUE_TOKEN }}
+          name: tls-get-observation-${{ github.run_id }}-${{ github.run_attempt }}
+          path: |
+            target/nextest/e2e-distributed/junit.xml
+            ${{ runner.temp }}/tls-get-nextest.log
+            ${{ runner.temp }}/tls-get-node-logs/internode-tls-*/node-0.log
+            ${{ runner.temp }}/tls-get-node-logs/internode-tls-*/node-1.log
+            ${{ runner.temp }}/tls-get-node-logs/internode-tls-*/node-2.log
+            ${{ runner.temp }}/tls-get-node-logs/internode-tls-*/node-3.log
+          retention-days: 7
+          if-no-files-found: warn
@@ -4075,6 +4075,7 @@ dependencies = [
 "rustfs-signer",
 "rustfs-tokio-tar",
 "rustfs-utils",
+ "rustix",
 "rustls",
 "s3s",
 "same-file",
@@ -4095,6 +4096,7 @@ dependencies = [
 "uuid",
 "walkdir",
 "zip",
+ "zeroize",
 "zstd 0.14.3",
 ]

@@ -143,8 +143,12 @@ rustls = { workspace = true, default-features = false, features = ["aws-lc-rs",
 russh = { workspace = true, features = ["serde"] }
 russh-sftp = { workspace = true }
 zip.workspace = true
+zeroize.workspace = true
 clap = { workspace = true, features = ["derive", "env"] }

+[target.'cfg(target_os = "linux")'.dependencies]
+rustix = { workspace = true, features = ["fs"] }
+
 [dev-dependencies]
 same-file.workspace = true
 aws-smithy-http-client = { workspace = true, default-features = false, features = ["test-util"] }
@@ -1589,10 +1589,29 @@ impl RustFSTestClusterEnvironment {
    /// Upgrade compatibility tests use this to initialize a cluster with a
    /// pinned previous release before replacing nodes with the workspace build.
    pub async fn start_with_binary(&mut self, binary_path: &Path) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        self.start_with_binary_and_callback(binary_path, |_| {}).await
+    }
+
+    #[cfg(target_os = "linux")]
+    pub(crate) async fn start_with_pid_callback(
+        &mut self,
+        on_pid: impl FnMut(u32),
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        let binary_path = rustfs_binary_path();
+        self.start_with_binary_and_callback(&binary_path, on_pid).await
+    }
+
+    async fn start_with_binary_and_callback(
+        &mut self,
+        binary_path: &Path,
+        mut on_pid: impl FnMut(u32),
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        let volumes_arg = self.build_volumes_arg();

        for node_idx in 0..self.nodes.len() {
            self.spawn_node(node_idx, binary_path, &volumes_arg)?;
+            let process = self.nodes[node_idx].process.as_ref().ok_or("spawned node has no process")?;
+            on_pid(process.id());
        }

        for (i, node) in self.nodes.iter().enumerate() {
@@ -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())
@@ -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();
@@ -329,7 +329,7 @@ zstd.workspace = true
 ed25519-dalek.workspace = true
 rustls = { workspace = true, default-features = false, features = ["aws-lc-rs", "logging", "tls12", "prefer-post-quantum", "std"] }
 rustls-pki-types = { workspace = true }
-rustix.workspace = true
+rustix = { workspace = true, features = ["fs"] }
 x509-parser = { workspace = true }
 subtle = { workspace = true }
 jiff = { workspace = true, features = ["serde"] }
@@ -1370,6 +1370,8 @@ struct GetObjectStreamingReader<R> {
    content_range: Option<String>,
    expected: usize,
    emitted: usize,
+    #[cfg(all(feature = "e2e-test-hooks", target_os = "linux"))]
+    e2e_observation: Option<super::get_e2e_observation::Observation>,
    timeout: Duration,
    timer: Option<Pin<Box<tokio::time::Sleep>>>,
    started: std::time::Instant,
@@ -1401,6 +1403,8 @@ impl<R> GetObjectStreamingReader<R> {
            content_range,
            expected,
            emitted: 0,
+            #[cfg(all(feature = "e2e-test-hooks", target_os = "linux"))]
+            e2e_observation: super::get_e2e_observation::Observation::start(request_id, expected),
            timeout,
            timer: None,
            started: std::time::Instant::now(),
@@ -1758,6 +1762,10 @@ impl<R: AsyncRead + Unpin> AsyncRead for GetObjectStreamingReader<R> {

 impl<R> Drop for GetObjectStreamingReader<R> {
    fn drop(&mut self) {
+        #[cfg(all(feature = "e2e-test-hooks", target_os = "linux"))]
+        if let Some(observation) = &mut self.e2e_observation {
+            observation.finish(self.emitted);
+        }
        if self.lifecycle.is_finished() {
            return;
        }
@@ -0,0 +1,498 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+//! Bounded, opt-in GET observations for a local distributed test.
+//!
+//! A line only establishes that the named reader event happened. A missing or
+//! truncated line is always unknown: the process may be killed, the channel
+//! may be full, or the writer may fail. Reader output is not client receipt.
+
+use hmac::{Hmac, Mac, digest::KeyInit};
+use sha2::Sha256;
+use std::io::{self, Write};
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, OnceLock};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+use tokio::io::AsyncReadExt;
+use tokio::sync::mpsc;
+use zeroize::Zeroizing;
+
+type TagMac = Hmac<Sha256>;
+const OBSERVED_SIZE: usize = 1024 * 1024 * 1024;
+const MAX_RECORDS: usize = 128;
+const CHANNEL_CAPACITY: usize = 64;
+const DIRECTORY_ENV: &str = "RUSTFS_E2E_TLS_GET_OBSERVATION_DIR";
+static SINK: OnceLock<Option<Arc<Sink>>> = OnceLock::new();
+
+struct Sink {
+    mac: TagMac,
+    sender: mpsc::Sender<Record>,
+    attempts: AtomicUsize,
+}
+
+#[derive(Clone, Copy)]
+enum Event {
+    Constructed,
+    Dropped,
+}
+
+struct Record {
+    pid: u32,
+    tag: [u8; 32],
+    event: Event,
+    expected: u64,
+    emitted: Option<u64>,
+    elapsed_ms: Option<u64>,
+    unix_ms: Option<u64>,
+    signature: [u8; 32],
+}
+
+impl Sink {
+    fn send(&self, mut record: Record) -> bool {
+        if self
+            .attempts
+            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| (n < MAX_RECORDS).then_some(n + 1))
+            .is_err()
+        {
+            return false;
+        }
+        record.pid = std::process::id();
+        let digest = record_mac(&self.mac, &record).finalize().into_bytes();
+        record.signature.copy_from_slice(&digest);
+        self.sender.try_send(record).is_ok()
+    }
+
+    fn tag(&self, request_id: &str) -> Option<[u8; 32]> {
+        let id = uuid::Uuid::parse_str(request_id).ok()?;
+        let mut mac = self.mac.clone();
+        mac.update(b"rustfs-get-observation-request-v1\0");
+        mac.update(id.as_bytes());
+        let digest = mac.finalize().into_bytes();
+        let mut tag = [0; 32];
+        tag.copy_from_slice(&digest);
+        Some(tag)
+    }
+}
+
+fn record_mac(mac: &TagMac, record: &Record) -> TagMac {
+    let mut mac = mac.clone();
+    mac.update(b"rustfs-get-observation-record-v1\0");
+    mac.update(&record.pid.to_be_bytes());
+    mac.update(&record.tag);
+    mac.update(&[match record.event {
+        Event::Constructed => 1,
+        Event::Dropped => 2,
+    }]);
+    mac.update(&record.expected.to_be_bytes());
+    for value in [record.emitted, record.elapsed_ms, record.unix_ms] {
+        mac.update(&[u8::from(value.is_some())]);
+        mac.update(&value.unwrap_or_default().to_be_bytes());
+    }
+    mac
+}
+
+/// Initialize once before accepting requests. Without the private directory the
+/// feature is inert. The key is sent over a private local socket and never
+/// placed in the process environment, a record, or diagnostic output.
+pub(crate) async fn init() {
+    let Some(directory) = std::env::var_os(DIRECTORY_ENV).map(PathBuf::from) else {
+        let _ = SINK.set(None);
+        return;
+    };
+    let result = tokio::time::timeout(Duration::from_secs(10), async {
+        let (sender, receiver) = tokio::sync::oneshot::channel();
+        let checked_directory = directory.clone();
+        std::thread::Builder::new()
+            .name("e2e-get-directory-check".into())
+            .spawn(move || {
+                let _ = sender.send(private_directory_identity(&checked_directory));
+            })
+            .ok()?;
+        let (directory_uid, parent_pid) = receiver.await.ok()??;
+        let mut stream = tokio::net::UnixStream::connect(directory.join("key.sock")).await.ok()?;
+        let peer = stream.peer_cred().ok()?;
+        if !peer_is_parent(peer.uid(), peer.pid().and_then(|pid| u32::try_from(pid).ok()), directory_uid, parent_pid) {
+            return None;
+        }
+        let mut key = Zeroizing::new([0u8; 64]);
+        stream.read_exact(&mut key[..]).await.ok()?;
+        if !key.iter().all(|byte| byte.is_ascii_hexdigit()) {
+            return None;
+        }
+        let mac = TagMac::new_from_slice(&key[..]).ok()?;
+        let (sender, receiver) = mpsc::channel(CHANNEL_CAPACITY);
+        std::thread::Builder::new()
+            .name("e2e-get-observation".into())
+            .spawn(move || consume(directory, receiver))
+            .ok()?;
+        Some(Arc::new(Sink {
+            mac,
+            sender,
+            attempts: AtomicUsize::new(0),
+        }))
+    })
+    .await
+    .ok()
+    .flatten();
+    let _ = SINK.set(result);
+}
+
+fn parent_pid(status: &str) -> Option<u32> {
+    status
+        .lines()
+        .find_map(|line| line.strip_prefix("PPid:")?.trim().parse().ok())
+}
+
+fn peer_is_parent(uid: u32, pid: Option<u32>, expected_uid: u32, expected_pid: u32) -> bool {
+    uid == expected_uid && pid == Some(expected_pid)
+}
+
+fn open_private_directory(path: &Path) -> Option<rustix::fd::OwnedFd> {
+    use rustix::fs::{CWD, FileType, Mode, OFlags, ResolveFlags, fstat, openat2};
+    let fd = openat2(
+        CWD,
+        path,
+        OFlags::PATH | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+        Mode::empty(),
+        ResolveFlags::NO_SYMLINKS,
+    )
+    .ok()?;
+    let stat = fstat(&fd).ok()?;
+    (FileType::from_raw_mode(stat.st_mode) == FileType::Directory && stat.st_mode & 0o077 == 0).then_some(fd)
+}
+
+fn private_directory_identity(path: &Path) -> Option<(u32, u32)> {
+    let directory = open_private_directory(path)?;
+    let uid = rustix::fs::fstat(&directory).ok()?.st_uid;
+    let status = std::fs::read_to_string("/proc/self/status").ok()?;
+    Some((uid, parent_pid(&status)?))
+}
+
+fn open_new_event_file(directory: &Path, pid: u32) -> Option<std::fs::File> {
+    let directory = open_private_directory(directory)?;
+    create_event_in_directory(&directory, pid)
+}
+
+fn create_event_in_directory(directory: &rustix::fd::OwnedFd, pid: u32) -> Option<std::fs::File> {
+    use rustix::fs::{FileType, Mode, OFlags, fstat, openat};
+    let name = format!("get-reader-{pid}.events");
+    let file = openat(
+        directory,
+        name.as_str(),
+        OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL | OFlags::NOFOLLOW | OFlags::NONBLOCK | OFlags::CLOEXEC,
+        Mode::from_raw_mode(0o600),
+    )
+    .ok()?;
+    let stat = fstat(&file).ok()?;
+    if FileType::from_raw_mode(stat.st_mode) != FileType::RegularFile || stat.st_nlink != 1 {
+        return None;
+    }
+    Some(file.into())
+}
+
+fn consume(directory: PathBuf, mut receiver: mpsc::Receiver<Record>) {
+    let Some(mut file) = open_new_event_file(&directory, std::process::id()) else {
+        return;
+    };
+    while let Some(record) = receiver.blocking_recv() {
+        let mut bytes = Vec::with_capacity(256);
+        if write_record(&mut bytes, &record).is_err() || file.write_all(&bytes).is_err() {
+            return;
+        }
+    }
+}
+
+fn write_record(writer: &mut impl Write, record: &Record) -> io::Result<()> {
+    let event = match record.event {
+        Event::Constructed => "constructed",
+        Event::Dropped => "dropped",
+    };
+    let tag = hex_simd::encode_to_string(record.tag, hex_simd::AsciiCase::Lower);
+    let signature = hex_simd::encode_to_string(record.signature, hex_simd::AsciiCase::Lower);
+    let value = |n: Option<u64>| n.map_or_else(|| "unknown".to_string(), |n| n.to_string());
+    writeln!(
+        writer,
+        "v1 {} {tag} {event} {} {} {} {} {signature}",
+        record.pid,
+        record.expected,
+        value(record.emitted),
+        value(record.elapsed_ms),
+        value(record.unix_ms)
+    )
+}
+
+fn unix_ms() -> Option<u64> {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .ok()
+        .and_then(|duration| u64::try_from(duration.as_millis()).ok())
+}
+
+pub(crate) struct Observation {
+    sink: Arc<Sink>,
+    tag: [u8; 32],
+    expected: u64,
+    emitted: Option<u64>,
+    started: Instant,
+}
+
+impl Observation {
+    pub(crate) fn start(request_id: &str, expected: usize) -> Option<Self> {
+        if expected != OBSERVED_SIZE {
+            return None;
+        }
+        Self::from_sink(SINK.get()?.as_ref()?.clone(), request_id, expected)
+    }
+
+    fn from_sink(sink: Arc<Sink>, request_id: &str, expected: usize) -> Option<Self> {
+        let tag = sink.tag(request_id)?;
+        let expected = u64::try_from(expected).ok()?;
+        let _ = sink.send(Record {
+            pid: 0,
+            tag,
+            event: Event::Constructed,
+            expected,
+            emitted: None,
+            elapsed_ms: Some(0),
+            unix_ms: unix_ms(),
+            signature: [0; 32],
+        });
+        Some(Self {
+            sink,
+            tag,
+            expected,
+            emitted: None,
+            started: Instant::now(),
+        })
+    }
+
+    pub(crate) fn finish(&mut self, emitted: usize) {
+        self.emitted = u64::try_from(emitted).ok();
+    }
+}
+
+impl Drop for Observation {
+    fn drop(&mut self) {
+        let _ = self.sink.send(Record {
+            pid: 0,
+            tag: self.tag,
+            event: Event::Dropped,
+            expected: self.expected,
+            emitted: self.emitted,
+            elapsed_ms: u64::try_from(self.started.elapsed().as_millis()).ok(),
+            unix_ms: unix_ms(),
+            signature: [0; 32],
+        });
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn sink(capacity: usize) -> (Arc<Sink>, mpsc::Receiver<Record>) {
+        let (sender, receiver) = mpsc::channel(capacity);
+        let mac = TagMac::new_from_slice(b"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef")
+            .expect("valid fixture key");
+        (
+            Arc::new(Sink {
+                mac,
+                sender,
+                attempts: AtomicUsize::new(0),
+            }),
+            receiver,
+        )
+    }
+
+    #[test]
+    fn event_writer_rejects_existing_fifo_symlinks_and_creates_a_regular_file() {
+        use rustix::fs::{CWD, Mode, mkfifoat};
+        use std::os::unix::fs::{PermissionsExt, symlink};
+
+        let directory = tempfile::Builder::new()
+            .permissions(std::fs::Permissions::from_mode(0o700))
+            .tempdir()
+            .expect("private fixture directory");
+        assert!(open_private_directory(directory.path()).is_some(), "private fixture must be admitted");
+        let file = open_new_event_file(directory.path(), 103).expect("new event file");
+        assert!(file.metadata().expect("event metadata").is_file());
+        let fifo = directory.path().join("get-reader-101.events");
+        mkfifoat(CWD, fifo.as_path(), Mode::from_raw_mode(0o600)).expect("FIFO fixture");
+        let owned_directory = directory.path().to_path_buf();
+        let (sender, receiver) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            let _ = sender.send(open_new_event_file(&owned_directory, 101).is_none());
+        });
+        assert!(
+            receiver
+                .recv_timeout(Duration::from_secs(1))
+                .expect("writer open must not block")
+        );
+
+        let target = directory.path().join("target");
+        std::fs::write(&target, b"unchanged").expect("target fixture");
+        symlink(&target, directory.path().join("get-reader-102.events")).expect("symlink fixture");
+        assert!(open_new_event_file(directory.path(), 102).is_none());
+        assert_eq!(std::fs::read(target).expect("target bytes"), b"unchanged");
+        let directory_link = directory.path().join("directory-link");
+        symlink(directory.path(), &directory_link).expect("directory symlink fixture");
+        assert!(open_new_event_file(&directory_link, 104).is_none());
+    }
+
+    #[test]
+    fn event_writer_remains_in_the_pinned_directory_after_path_replacement() {
+        use std::os::unix::fs::DirBuilderExt;
+
+        let root = tempfile::tempdir().expect("private fixture root");
+        let directory = root.path().join("events");
+        let moved = root.path().join("moved");
+        std::fs::DirBuilder::new()
+            .mode(0o700)
+            .create(&directory)
+            .expect("event directory");
+        let pinned = open_private_directory(&directory).expect("pin event directory");
+        std::fs::rename(&directory, &moved).expect("move pinned directory");
+        std::fs::DirBuilder::new()
+            .mode(0o700)
+            .create(&directory)
+            .expect("replacement directory");
+        let file = create_event_in_directory(&pinned, 123).expect("write in pinned directory");
+        assert!(file.metadata().expect("event metadata").is_file());
+        assert!(moved.join("get-reader-123.events").is_file());
+        assert!(!directory.join("get-reader-123.events").exists());
+    }
+
+    #[test]
+    fn parent_peer_requires_both_kernel_pid_and_uid() {
+        assert!(peer_is_parent(1000, Some(41), 1000, 41));
+        assert!(!peer_is_parent(1000, Some(42), 1000, 41));
+        assert!(!peer_is_parent(1000, None, 1000, 41));
+        assert!(!peer_is_parent(1001, Some(41), 1000, 41));
+        assert_eq!(parent_pid("Name:\trustfs\nPPid:\t41\n"), Some(41));
+        assert_eq!(parent_pid("Name:\trustfs\n"), None);
+    }
+
+    #[tokio::test]
+    async fn parent_check_reads_kernel_socket_credentials() {
+        let (first, _second) = tokio::net::UnixStream::pair().expect("local socket pair");
+        let peer = first.peer_cred().expect("peer credentials");
+        let pid = peer.pid().and_then(|pid| u32::try_from(pid).ok());
+        let own_pid = std::process::id();
+        assert!(peer_is_parent(peer.uid(), pid, peer.uid(), own_pid));
+        assert!(!peer_is_parent(peer.uid(), pid, peer.uid(), own_pid.saturating_add(1)));
+    }
+
+    #[test]
+    fn authenticated_record_matches_independent_fixture() {
+        let (sink, _receiver) = sink(1);
+        let tag = hex_simd::decode_to_vec("636a632459fe3a2e3d23493ca06429fc88e6ebf79aa11352a1d7b39cefe401d0")
+            .expect("fixture tag")
+            .try_into()
+            .expect("32-byte tag");
+        let record = Record {
+            pid: 4242,
+            tag,
+            event: Event::Constructed,
+            expected: 1024 * 1024 * 1024,
+            emitted: None,
+            elapsed_ms: Some(0),
+            unix_ms: Some(1_700_000_000_000),
+            signature: [0; 32],
+        };
+        let digest = record_mac(&sink.mac, &record).finalize().into_bytes();
+        assert_eq!(
+            hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower),
+            "d19307fd28aa41b5177b014ee5d725fb19a20523c54b985c33cd3eccdced8bda"
+        );
+    }
+
+    #[test]
+    fn records_are_redacted_and_disconnected_receiver_is_nonfatal() {
+        let (sink, mut receiver) = sink(4);
+        let id = "b8172a15-a428-4b42-9cef-1f4a7b96874e";
+        let mut observation = Observation::from_sink(sink.clone(), id, 1024).expect("valid UUID");
+        observation.finish(42);
+        drop(observation);
+        let first = receiver.try_recv().expect("constructed");
+        let second = receiver.try_recv().expect("dropped");
+        assert_eq!(first.tag, second.tag);
+        assert_eq!(second.emitted, Some(42));
+        let mut bytes = Vec::new();
+        write_record(&mut bytes, &first).expect("write constructed");
+        write_record(&mut bytes, &second).expect("write dropped");
+        let text = String::from_utf8(bytes).expect("ASCII record");
+        assert!(!text.contains(id));
+        assert!(!text.contains("0123456789abcdef"));
+        assert!(text.contains("636a632459fe3a2e3d23493ca06429fc88e6ebf79aa11352a1d7b39cefe401d0"));
+        assert!(Observation::from_sink(sink.clone(), "bad request id", 1024).is_none());
+        drop(receiver);
+        let mut observation = Observation::from_sink(sink, id, 1024).expect("closed sink is nonfatal");
+        observation.finish(5);
+    }
+
+    #[test]
+    fn full_queue_and_panic_leave_missing_values_unknown() {
+        let (sink, mut receiver) = sink(1);
+        let id = "b8172a15-a428-4b42-9cef-1f4a7b96874e";
+        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            let _observation = Observation::from_sink(sink, id, 1024).expect("valid UUID");
+            panic!("fixture");
+        }));
+        assert!(result.is_err());
+        assert!(matches!(receiver.try_recv().expect("constructed").event, Event::Constructed));
+        assert!(receiver.try_recv().is_err(), "dropped event was lost when the queue was full");
+    }
+
+    #[test]
+    fn unwind_drop_does_not_claim_a_byte_count() {
+        let (sink, mut receiver) = sink(2);
+        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            let _observation = Observation::from_sink(sink, "b8172a15-a428-4b42-9cef-1f4a7b96874e", 1024).expect("valid UUID");
+            panic!("fixture");
+        }));
+        assert!(result.is_err());
+        let _constructed = receiver.try_recv().expect("constructed");
+        let dropped = receiver.try_recv().expect("drop during unwind");
+        assert!(matches!(dropped.event, Event::Dropped));
+        assert!(dropped.emitted.is_none());
+    }
+
+    #[test]
+    fn attempted_records_are_capped_even_with_a_live_receiver() {
+        let (sink, mut receiver) = sink(MAX_RECORDS + 1);
+        for _ in 0..MAX_RECORDS + 1 {
+            let _ = sink.send(Record {
+                pid: 0,
+                tag: [0; 32],
+                event: Event::Constructed,
+                expected: 1,
+                emitted: None,
+                elapsed_ms: None,
+                unix_ms: None,
+                signature: [0; 32],
+            });
+        }
+        for _ in 0..MAX_RECORDS {
+            receiver.try_recv().expect("within the cap");
+        }
+        assert!(receiver.try_recv().is_err());
+    }
+
+    #[test]
+    fn writer_failure_is_reported_without_record_contents() {
+        struct FailingWriter;
+        impl Write for FailingWriter {
+            fn write(&mut self, _bytes: &[u8]) -> io::Result<usize> {
+                Err(io::Error::other("injected writer failure"))
+            }
+            fn flush(&mut self) -> io::Result<()> {
+                Ok(())
+            }
+        }
+        let (sink, mut receiver) = sink(1);
+        let _guard = Observation::from_sink(sink, "b8172a15-a428-4b42-9cef-1f4a7b96874e", 1024).expect("valid UUID");
+        let record = receiver.try_recv().expect("constructed");
+        assert!(write_record(&mut FailingWriter, &record).is_err());
+    }
+}
@@ -211,6 +211,8 @@ mod copy;
 mod delete;
 mod extract;
 mod get;
+#[cfg(all(feature = "e2e-test-hooks", target_os = "linux"))]
+pub(crate) mod get_e2e_observation;
 mod head;
 mod internal_put;
 mod on_demand_migration_put;
@@ -1980,6 +1980,9 @@ fn print_license_report(report: &crate::connect::LicenseReport) -> Result<()> {

 #[instrument(skip(config))]
 async fn run(config: Config) -> Result<()> {
+    #[cfg(all(feature = "e2e-test-hooks", target_os = "linux"))]
+    crate::app::object::get_e2e_observation::init().await;
+
    // Single-instance startup threads the process bootstrap context through
    // the storage path explicitly (Phase 5 follow-up, backlog#1052); a future
    // multi-instance server constructs its own context here instead.
Author	SHA1	Message	Date
overtrue	05c9c7a87b	ci: isolate TLS diagnostic report environment	2026-10-28 11:32:05 +08:00
overtrue	0339b4f881	test: create private TLS observation fixture atomically	2026-10-28 10:07:29 +08:00
overtrue	7d3f077072	test: capture bounded TLS read observations	2026-10-28 09:25:01 +08:00
overtrue	fedc966612	Capture concurrent TLS read failure progress	2026-10-28 05:36:17 +08:00
overtrue	80f80afc62	Merge commit 'f96c94f95056697be2e0257320306592946642c5' into overtrue/scanner-heal-identity-scope	2026-10-28 03:00:56 +08:00
overtrue	f691431400	Merge remote-tracking branch 'origin/main' into overtrue/scanner-heal-identity-scope	2026-10-28 01:07:54 +08:00
overtrue	e776986d62	Merge remote-tracking branch 'origin/main' into overtrue/scanner-heal-identity-scope	2026-10-28 00:11:34 +08:00
overtrue	fe8b36cd79	Merge remote-tracking branch 'origin/main' into overtrue/scanner-heal-identity-scope	2026-10-27 23:53:55 +08:00
overtrue	536f5bf96c	test(scanner): simplify identity case types	2026-10-27 23:53:55 +08:00
overtrue	7739a37978	fix(scanner): distinguish missing heal identity fields	2026-10-27 22:39:38 +08:00