ci: prepare asynchronous isolation preflight

.github/workflows/windows-filesystem.yml

@@ -1,224 +1,259 @@
-# 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/licenses/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.
-
-name: Windows Filesystem Tests
+# Temporary, manual-only verification of the test-only isolation fixture.
+# The exporter is in memory; no OTLP wire serialization is exercised.
+name: TLS diagnostic async isolation preflight
 
 on:
-  push:
-    branches: [ main ]
-    paths:
-      - "crates/ecstore/src/disk/**"
-      - "crates/ecstore/src/layout/**"
-      - "crates/ecstore/src/store/init_format.rs"
-      - "crates/ecstore/src/set_disk/ops/locking.rs"
-      - "crates/ecstore/src/object_api/write_commit_context.rs"
-      - "crates/ecstore/src/bucket/utils.rs"
-      - "crates/ecstore/Cargo.toml"
-      - "crates/utils/src/dunce.rs"
-      - "crates/utils/src/os/**"
-      - "crates/utils/Cargo.toml"
-      - "crates/utils/src/path.rs"
-      - "rustfs/build.rs"
-      - "rustfs/Cargo.toml"
-      - "rustfs/src/main.rs"
-      - "rustfs/src/server/runtime.rs"
-      - "rustfs/src/startup_*.rs"
-      - "Cargo.toml"
-      - "Cargo.lock"
-      - ".github/actions/setup/**"
-      - ".github/workflows/windows-filesystem.yml"
-  pull_request:
-    branches: [ main ]
-    paths:
-      - "crates/ecstore/src/disk/**"
-      - "crates/ecstore/src/layout/**"
-      - "crates/ecstore/src/store/init_format.rs"
-      - "crates/ecstore/src/set_disk/ops/locking.rs"
-      - "crates/ecstore/src/object_api/write_commit_context.rs"
-      - "crates/ecstore/src/bucket/utils.rs"
-      - "crates/ecstore/Cargo.toml"
-      - "crates/utils/src/dunce.rs"
-      - "crates/utils/src/os/**"
-      - "crates/utils/Cargo.toml"
-      - "crates/utils/src/path.rs"
-      - "rustfs/build.rs"
-      - "rustfs/Cargo.toml"
-      - "rustfs/src/main.rs"
-      - "rustfs/src/server/runtime.rs"
-      - "rustfs/src/startup_*.rs"
-      - "Cargo.toml"
-      - "Cargo.lock"
-      - ".github/actions/setup/**"
-      - ".github/workflows/windows-filesystem.yml"
   workflow_dispatch:
 
 permissions:
   contents: read
 
-concurrency:
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
-  cancel-in-progress: true
-
-env:
-  CARGO_TERM_COLOR: always
-  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
-  RUST_BACKTRACE: 1
-
 jobs:
-  rename-safety:
-    name: Rename Safety
-    runs-on: windows-latest
-    timeout-minutes: 60
+  isolation:
+    runs-on: ubuntu-latest
+    timeout-minutes: 120
+    env:
+      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+      CARGO_INCREMENTAL: "0"
+      CARGO_BUILD_JOBS: "2"
+      CARGO_PROFILE_DEV_DEBUG: "0"
+      CARGO_PROFILE_TEST_DEBUG: "0"
+      CARGO_TERM_COLOR: never
+      RUST_BACKTRACE: "0"
+      CARGO_TARGET_DIR: ${{ github.workspace }}/target
+      EXPECTED_REVISION: 6010f8cd2d0dab27e52a98d305afd11926df56dd
+      EXPECTED_TREE: cf952d55d16f7c159fc6c7b0dfa2ec35a6df38a9
     steps:
-      - name: Checkout repository
+      - name: Checkout fixed source
+        id: source
         uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
         with:
+          ref: 6010f8cd2d0dab27e52a98d305afd11926df56dd
+          clean: false
           persist-credentials: false
 
-      - name: Setup Rust environment
-        uses: ./.github/actions/setup
-        with:
-          rust-version: stable
-          cache-shared-key: build-x86_64-pc-windows-msvc
-          cache-save-if: 'false'
-          install-build-packaging-tools: 'false'
-          install-test-tools: 'false'
-
-      - name: Check production Windows dependencies
-        shell: pwsh
-        run: cargo check -p rustfs-ecstore --lib
-
-      - name: Test guarded rename publication
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib rename_all_ -- --nocapture
-
-      - name: Test multipart parent durability barriers
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib multipart_owned_record_parent_barriers_ -- --nocapture
-
-      - name: Test Windows path conversion
-        shell: pwsh
-        run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
-
-      - name: Test Windows volume and disk-space probes
-        shell: pwsh
-        run: |
-          # Map a drive letter to the temp drive's administrative share so the
-          # remote-volume probe also covers mapped network drives.
-          $drive = (Split-Path -Qualifier ([System.IO.Path]::GetTempPath())).TrimEnd(':')
-          net use R: "\\localhost\$drive`$" /persistent:no
-          if ($LASTEXITCODE -eq 0) {
-            $env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE = 'R:'
-          } else {
-            Write-Output "::warning::could not map R: to \\localhost\$drive`$; mapped-drive test will skip"
-          }
-          cargo test -p rustfs-utils --features os --lib os:: -- --nocapture
-          $status = $LASTEXITCODE
-          if ($env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE) { net use R: /delete /y | Out-Null }
-          exit $status
-
-      - name: Test Windows handle guards
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib windows_ -- --nocapture
-
-      - name: Test endpoint path parsing
-        shell: pwsh
-        run: "cargo test -p rustfs-ecstore --lib layout::endpoint:: -- --nocapture"
-
-      - name: Test startup temporary-directory cleanup
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib cleanup_tmp_on_startup_ -- --nocapture
-
-      - name: Test object key Windows admission
-        shell: pwsh
-        run: "cargo test -p rustfs-ecstore --lib bucket::utils:: -- --nocapture"
-
-      - name: Test Windows file-name rules
-        shell: pwsh
-        run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
-
-      - name: Test fresh format publication
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib fresh_format_load_initializes_all_disks -- --nocapture
-
-  # The native Windows main thread only gets the 1 MiB PE stack reserve, while
-  # Linux and macOS hand out 8 MiB. Startup runs the whole root future on the
-  # thread that calls `block_on`, so a debug server once overflowed before it
-  # listened on Windows only. Start a real server to keep that covered.
-  startup-smoke:
-    name: Startup Smoke
-    runs-on: windows-latest
-    timeout-minutes: 90
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
-        with:
-          persist-credentials: false
-
-      - name: Setup Rust environment
-        uses: ./.github/actions/setup
-        with:
-          rust-version: stable
-          cache-shared-key: build-x86_64-pc-windows-msvc
-          cache-save-if: 'false'
-          install-build-packaging-tools: 'false'
-          install-test-tools: 'false'
-
-      - name: Build debug server binary
-        shell: pwsh
-        run: cargo build --locked -p rustfs --bin rustfs
-
-      - name: Print version
-        shell: pwsh
-        run: |
-          & .\target\debug\rustfs.exe --version
-          if ($LASTEXITCODE -ne 0) { throw "rustfs --version exited with $LASTEXITCODE" }
-
-      - name: Start server and wait for readiness
-        shell: pwsh
+      - name: Verify source and prepare private output
+        id: identity
         env:
-          RUSTFS_ACCESS_KEY: rustfsadmin
-          RUSTFS_SECRET_KEY: rustfsadmin
+          WORKFLOW_REVISION: ${{ github.sha }}
         run: |
-          $volume = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke'
-          $stdout = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.out.log'
-          $stderr = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.err.log'
-          New-Item -ItemType Directory -Force -Path $volume | Out-Null
-          $server = Start-Process -FilePath .\target\debug\rustfs.exe `
-            -ArgumentList @('--address', '127.0.0.1:9000', "`"$volume`"") `
-            -RedirectStandardOutput $stdout -RedirectStandardError $stderr `
-            -NoNewWindow -PassThru
-          # Cache the process handle now; otherwise ExitCode reads back empty.
-          $null = $server.Handle
-          $ready = $false
-          try {
-            for ($i = 0; $i -lt 180; $i++) {
-              if ($server.HasExited) { break }
-              try {
-                $response = Invoke-WebRequest -Uri 'http://127.0.0.1:9000/health/ready' -UseBasicParsing -NoProxy -TimeoutSec 2
-                if ($response.StatusCode -eq 200) { $ready = $true; break }
-              } catch {}
-              Start-Sleep -Seconds 1
-            }
-          } finally {
-            if (-not $server.HasExited) { Stop-Process -Id $server.Id -Force }
-            $server.WaitForExit()
-            Write-Host '--- server stdout (tail) ---'
-            Get-Content $stdout -Tail 100 -ErrorAction SilentlyContinue
-            Write-Host '--- server stderr (tail) ---'
-            Get-Content $stderr -Tail 100 -ErrorAction SilentlyContinue
-          }
-          if (-not $ready) {
-            throw ("rustfs did not report ready on /health/ready (exit code: {0})" -f $server.ExitCode)
+          set -euo pipefail
+          test "$(git rev-parse HEAD)" = "$EXPECTED_REVISION"
+          test "$(git rev-parse 'HEAD^{tree}')" = "$EXPECTED_TREE"
+          test -z "$(git status --porcelain=v1 --untracked-files=all)"
+          umask 077
+          mkdir "$RUNNER_TEMP/tls-async-private"
+          mkdir "$RUNNER_TEMP/tls-async-evidence"
+          python3 - <<'PY'
+          import hashlib
+          import json
+          import os
+          import sys
+          from pathlib import Path
+
+          sys.excepthook = lambda *_: print('Source identity unavailable; raw output withheld.', file=sys.stderr)
+          source = Path('crates/obs/src/telemetry/isolation_probe_tests.rs').read_bytes()
+          identity = {
+              'source_revision': os.environ['EXPECTED_REVISION'],
+              'source_tree': os.environ['EXPECTED_TREE'],
+              'workflow_revision': os.environ['WORKFLOW_REVISION'],
+              'test_sha256': hashlib.sha256(source).hexdigest(),
           }
+          output = Path(os.environ['RUNNER_TEMP'], 'tls-async-evidence', 'identity.json')
+          output.write_text(json.dumps(identity, sort_keys=True) + '\n')
+          PY
+
+      - 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
+        id: rust
+        uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
+        with:
+          toolchain: stable
+
+      - name: Record toolchain identity
+        id: toolchain
+        run: |
+          python3 - <<'PY'
+          import json
+          import os
+          import re
+          import subprocess
+          import sys
+          from pathlib import Path
+
+          sys.excepthook = lambda *_: print('Toolchain identity unavailable; raw output withheld.', file=sys.stderr)
+          versions = {}
+          for program in ('rustc', 'cargo'):
+              version = subprocess.run([program, '--version'], check=True, capture_output=True, text=True).stdout.strip()
+              if not re.fullmatch(program + r' [a-zA-Z0-9+(). _-]{1,120}', version):
+                  raise SystemExit('Unexpected toolchain version; raw output withheld.')
+              versions[program] = version
+          output = Path(os.environ['RUNNER_TEMP'], 'tls-async-evidence', 'toolchain.json')
+          output.write_text(json.dumps(versions, sort_keys=True) + '\n')
+          PY
+
+      - name: Compile the library test binary
+        id: compile
+        run: |
+          set -euo pipefail
+          if ! cargo test --locked -p rustfs-obs --lib --no-run --message-format=json >"$RUNNER_TEMP/tls-async-private/build.jsonl" 2>"$RUNNER_TEMP/tls-async-private/build.log"; then
+            echo "The library test binary did not compile; raw output withheld."
+            exit 1
+          fi
+
+      - name: Run the six named isolation tests
+        id: assertions
+        run: |
+          set -euo pipefail
+          test "$(git rev-parse HEAD)" = "$EXPECTED_REVISION"
+          test "$(git rev-parse 'HEAD^{tree}')" = "$EXPECTED_TREE"
+          test -z "$(git status --porcelain=v1 --untracked-files=all)"
+          python3 - <<'PY'
+          import hashlib
+          import json
+          import os
+          import subprocess
+          import sys
+          from pathlib import Path
+
+          sys.excepthook = lambda *_: print('Preflight evidence unavailable; raw output withheld.', file=sys.stderr)
+          prefix = 'telemetry::isolation_probe_tests::async_isolation::'
+          cases = (
+              ('threads', prefix + 'separate_threads_keep_their_contexts_across_async_suspension'),
+              ('shared_runtime', prefix + 'shared_runtime_reenters_outer_context_after_each_await'),
+              ('one_step_control', prefix + 'either_isolation_step_alone_leaves_ambient_context'),
+              ('filters_and_resource', prefix + 'filters_and_resources_do_not_cross_dispatchers'),
+              ('cancel_drop', prefix + 'cancellation_drop_emits_in_isolation_and_restores_context'),
+              ('panic_restore', prefix + 'panic_during_sink_write_restores_the_outer_context'),
+          )
+          root = Path(os.environ['RUNNER_TEMP'])
+          private = root / 'tls-async-private'
+          evidence = root / 'tls-async-evidence'
+          build = private / 'build.jsonl'
+          if not build.is_file() or build.stat().st_size > 256 * 1024 * 1024:
+              raise SystemExit('Build output unavailable or too large; raw output withheld.')
+          binaries = set()
+          with build.open(encoding='utf-8') as lines:
+              for line in lines:
+                  if len(line) > 2 * 1024 * 1024:
+                      raise SystemExit('Build output record too large; raw output withheld.')
+                  try:
+                      value = json.loads(line)
+                  except ValueError:
+                      raise SystemExit('Build output was not valid JSON; raw output withheld.')
+                  if value.get('reason') != 'compiler-artifact':
+                      continue
+                  target = value.get('target', {})
+                  profile = value.get('profile', {})
+                  executable = value.get('executable')
+                  if (target.get('name') == 'rustfs_obs' and 'lib' in target.get('kind', [])
+                          and profile.get('test') is True and isinstance(executable, str)):
+                      binaries.add(executable)
+          if len(binaries) != 1:
+              raise SystemExit('Expected one library test binary; raw output withheld.')
+          binary = Path(next(iter(binaries))).resolve(strict=True)
+          target_root = Path(os.environ['CARGO_TARGET_DIR']).resolve(strict=True)
+          if not binary.is_file() or not binary.is_relative_to(target_root):
+              raise SystemExit('Test binary identity was unavailable; raw output withheld.')
+
+          def digest(path):
+              hasher = hashlib.sha256()
+              with path.open('rb') as handle:
+                  for block in iter(lambda: handle.read(1024 * 1024), b''):
+                      hasher.update(block)
+              return hasher.hexdigest()
+
+          binary_sha256 = digest(binary)
+          binary_size = binary.stat().st_size
+          results = []
+          for index, (label, name) in enumerate(cases):
+              output = private / f'test-{index}.log'
+              command = ['cargo', 'test', '--offline', '--locked', '-p', 'rustfs-obs', '--lib', name,
+                         '--', '--exact', '--color', 'never']
+              try:
+                  with output.open('wb') as handle:
+                      completed = subprocess.run(command, stdout=handle, stderr=subprocess.STDOUT, timeout=120, check=False)
+              except subprocess.TimeoutExpired:
+                  results.append({'case': label, 'state': 'timeout', 'exit_code': None, 'selected': False, 'passed': False})
+                  break
+              if not output.is_file():
+                  results.append({'case': label, 'state': 'missing', 'exit_code': completed.returncode, 'selected': False, 'passed': False})
+                  continue
+              if output.stat().st_size > 16 * 1024 * 1024:
+                  results.append({'case': label, 'state': 'truncated', 'exit_code': completed.returncode, 'selected': False, 'passed': False})
+                  continue
+              lines = output.read_text(encoding='utf-8', errors='replace').splitlines()
+              selected = lines.count('running 1 test') == 1
+              passed = lines.count(f'test {name} ... ok') == 1
+              results.append({'case': label, 'state': 'available', 'exit_code': completed.returncode,
+                              'selected': selected, 'passed': passed})
+          binary_unchanged = binary.is_file() and binary.stat().st_size == binary_size and digest(binary) == binary_sha256
+          accepted = (len(results) == len(cases) and binary_unchanged
+                      and all(item['state'] == 'available' and item['exit_code'] == 0
+                              and item['selected'] and item['passed'] for item in results))
+          report = {'binary_sha256': binary_sha256, 'binary_size': binary_size, 'binary_unchanged': binary_unchanged,
+                    'expected_cases': len(cases), 'cases': results, 'expectation_met': accepted}
+          (evidence / 'results.json').write_text(json.dumps(report, sort_keys=True) + '\n')
+          if not accepted:
+              raise SystemExit('Isolation assertions did not all pass; raw output withheld.')
+          print('All six exact isolation tests passed.')
+          PY
+          test "$(git rev-parse HEAD)" = "$EXPECTED_REVISION"
+          test "$(git rev-parse 'HEAD^{tree}')" = "$EXPECTED_TREE"
+          test -z "$(git status --porcelain=v1 --untracked-files=all)"
+
+      - name: Record stage outcomes
+        if: always()
+        env:
+          SOURCE: ${{ steps.source.outcome }}
+          IDENTITY: ${{ steps.identity.outcome }}
+          DEPENDENCIES: ${{ steps.dependencies.outcome }}
+          PROTOC: ${{ steps.protoc.outcome }}
+          FLATC: ${{ steps.flatc.outcome }}
+          RUST: ${{ steps.rust.outcome }}
+          TOOLCHAIN: ${{ steps.toolchain.outcome }}
+          COMPILE: ${{ steps.compile.outcome }}
+          ASSERTIONS: ${{ steps.assertions.outcome }}
+        run: |
+          umask 077
+          mkdir -p "$RUNNER_TEMP/tls-async-evidence"
+          python3 - <<'PY'
+          import json
+          import os
+          import sys
+          from pathlib import Path
+
+          sys.excepthook = lambda *_: print('Stage evidence unavailable; raw output withheld.', file=sys.stderr)
+          names = ('SOURCE', 'IDENTITY', 'DEPENDENCIES', 'PROTOC', 'FLATC', 'RUST', 'TOOLCHAIN', 'COMPILE', 'ASSERTIONS')
+          allowed = {'success', 'failure', 'cancelled', 'skipped'}
+          stages = {name.lower(): os.environ.get(name) if os.environ.get(name) in allowed else 'missing' for name in names}
+          output = Path(os.environ['RUNNER_TEMP'], 'tls-async-evidence', 'stages.json')
+          output.write_text(json.dumps(stages, sort_keys=True) + '\n')
+          PY
+
+      - name: Upload sanitized evidence
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: tls-async-isolation-${{ github.run_id }}-${{ github.run_attempt }}
+          path: ${{ runner.temp }}/tls-async-evidence/
+          retention-days: 7
+          if-no-files-found: error

crates/obs/src/telemetry/isolation_probe_tests.rs

@@ -0,0 +1,680 @@
+// 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/licenses/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.
+
+//! Test-only diagnostic isolation prototype. Nothing here is a GET hook.
+
+use super::local::build_json_log_layer;
+use opentelemetry::{
+    Context, InstrumentationScope, KeyValue,
+    baggage::BaggageExt,
+    logs::AnyValue,
+    trace::{SpanContext, SpanId, TraceContextExt, TraceFlags, TraceId, TraceState},
+};
+use opentelemetry_appender_tracing::layer::{OpenTelemetryTracingBridge, TracingSpanAttributes};
+use opentelemetry_sdk::{
+    Resource,
+    error::OTelSdkResult,
+    logs::{LogBatch, LogExporter, SdkLogRecord, SdkLoggerProvider},
+};
+use serde_json::Value;
+use std::{
+    collections::BTreeSet,
+    io::{self, Write},
+    sync::{Arc, Mutex},
+};
+use tracing::{Dispatch, dispatcher};
+use tracing_subscriber::{fmt::format::FmtSpan, layer::SubscriberExt};
+
+const TARGET: &str = "rustfs_isolation_probe_test";
+const FAKE_URI: &str = "/synthetic/private-object";
+const FAKE_AUTH: &str = "synthetic-private-header";
+
+#[derive(Clone, Debug, Default)]
+struct MemoryWriter(Arc<Mutex<Vec<u8>>>);
+
+impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for MemoryWriter {
+    type Writer = Self;
+
+    fn make_writer(&'a self) -> Self::Writer {
+        self.clone()
+    }
+}
+
+impl Write for MemoryWriter {
+    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+        self.0.lock().expect("capture JSON output").extend_from_slice(bytes);
+        Ok(bytes.len())
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+#[derive(Clone, Debug, Default)]
+struct MemoryExporter {
+    records: Arc<Mutex<Vec<(SdkLogRecord, InstrumentationScope)>>>,
+    resource: Arc<Mutex<Option<Resource>>>,
+}
+
+impl LogExporter for MemoryExporter {
+    async fn export(&self, batch: LogBatch<'_>) -> OTelSdkResult {
+        self.records
+            .lock()
+            .expect("capture exporter records")
+            .extend(batch.iter().map(|(record, scope)| (record.clone(), scope.clone())));
+        Ok(())
+    }
+
+    fn set_resource(&mut self, resource: &Resource) {
+        *self.resource.lock().expect("capture exporter resource") = Some(resource.clone());
+    }
+}
+
+struct Sinks {
+    dispatch: Dispatch,
+    json: MemoryWriter,
+    exporter: MemoryExporter,
+    _provider: SdkLoggerProvider,
+}
+
+impl Sinks {
+    fn new(span_list: bool, enrich_spans: bool) -> Self {
+        let json = MemoryWriter::default();
+        let exporter = MemoryExporter::default();
+        let provider = SdkLoggerProvider::builder()
+            .with_resource(Resource::builder_empty().build())
+            .with_simple_exporter(exporter.clone())
+            .build();
+        let bridge = OpenTelemetryTracingBridge::builder(&provider);
+        let bridge = if enrich_spans {
+            bridge.with_tracing_span_attributes(TracingSpanAttributes::all()).build()
+        } else {
+            bridge.build()
+        };
+        let subscriber = tracing_subscriber::registry()
+            .with(build_json_log_layer(json.clone(), false, FmtSpan::NONE, span_list))
+            .with(bridge);
+        Self {
+            dispatch: Dispatch::new(subscriber),
+            json,
+            exporter,
+            _provider: provider,
+        }
+    }
+
+    fn json_records(&self) -> Vec<Value> {
+        let bytes = self.json.0.lock().expect("capture JSON output").clone();
+        let text = String::from_utf8(bytes).expect("JSON output is UTF-8");
+        text.lines()
+            .map(|line| serde_json::from_str(line).expect("each captured record is JSON"))
+            .collect()
+    }
+
+    fn exported_records(&self) -> Vec<(SdkLogRecord, InstrumentationScope)> {
+        self.exporter.records.lock().expect("capture exporter records").clone()
+    }
+}
+
+fn fake_ambient_context() -> Context {
+    Context::new()
+        .with_remote_span_context(SpanContext::new(
+            TraceId::from(17),
+            SpanId::from(19),
+            TraceFlags::SAMPLED,
+            true,
+            TraceState::default(),
+        ))
+        .with_baggage([KeyValue::new("synthetic-secret", FAKE_AUTH)])
+}
+
+fn emit_root_event(phase: u64) {
+    tracing::info!(target: TARGET, parent: None, expected = 1024_u64, emitted = 64_u64, phase, "diagnostic fixture");
+}
+
+// This prototype is synchronous and must use a fresh, span-free dispatcher.
+fn emit_candidate(dispatch: &Dispatch, phase: u64) {
+    let _context_guard = Context::new().attach();
+    dispatcher::with_default(dispatch, || {
+        assert!(
+            Context::current().baggage().get("synthetic-secret").is_none(),
+            "ambient baggage must be hidden"
+        );
+        assert!(!Context::current().has_active_span(), "ambient OTEL span must be hidden");
+        emit_root_event(phase);
+    });
+}
+
+#[test]
+fn parent_none_alone_keeps_json_parent_and_otel_thread_context() {
+    let sinks = Sinks::new(true, true);
+    let _context_guard = fake_ambient_context().attach();
+
+    dispatcher::with_default(&sinks.dispatch, || {
+        let span = tracing::info_span!("http-request", uri = FAKE_URI, authorization = FAKE_AUTH, request_id = "synthetic-id");
+        let _span_guard = span.enter();
+        emit_root_event(0);
+    });
+
+    let json = sinks.json_records();
+    let logs = sinks.exported_records();
+    assert_eq!(json.len(), 1, "the control event must be captured");
+    assert_eq!(logs.len(), 1, "the control event must reach the exporter");
+    assert_eq!(json[0]["span"]["uri"].as_str(), Some(FAKE_URI));
+    assert_eq!(json[0]["span"]["authorization"].as_str(), Some(FAKE_AUTH));
+    let context = logs[0].0.trace_context().expect("ambient OTEL trace context is present");
+    assert_eq!(context.trace_id, TraceId::from(17));
+    assert_eq!(context.span_id, SpanId::from(19));
+}
+
+#[test]
+fn diagnostic_fixture_is_detached_for_both_sinks_and_restores_context() {
+    for span_list in [false, true] {
+        for enrich_spans in [false, true] {
+            let outer = Sinks::new(span_list, enrich_spans);
+            let isolated = Sinks::new(span_list, enrich_spans);
+            let _context_guard = fake_ambient_context().attach();
+
+            dispatcher::with_default(&outer.dispatch, || {
+                let span =
+                    tracing::info_span!("http-request", uri = FAKE_URI, authorization = FAKE_AUTH, request_id = "synthetic-id");
+                let _span_guard = span.enter();
+                for phase in [0, 1] {
+                    emit_candidate(&isolated.dispatch, phase);
+                    assert_eq!(tracing::Span::current().id(), span.id(), "restore the outer tracing span");
+                    assert_eq!(Context::current().span().span_context().trace_id(), TraceId::from(17));
+                    assert!(Context::current().baggage().get("synthetic-secret").is_some());
+                }
+            });
+
+            assert!(outer.json_records().is_empty(), "diagnostic events must not reach the outer JSON sink");
+            assert!(outer.exported_records().is_empty(), "diagnostic events must not reach the outer exporter");
+            let json = isolated.json_records();
+            let logs = isolated.exported_records();
+            assert_eq!(json.len(), 2, "both diagnostic events must reach the isolated JSON sink");
+            assert_eq!(logs.len(), 2, "both diagnostic events must reach the isolated exporter");
+
+            let json_keys: BTreeSet<&str> = [
+                "timestamp",
+                "level",
+                "target",
+                "filename",
+                "line_number",
+                "threadName",
+                "threadId",
+                "message",
+                "expected",
+                "emitted",
+                "phase",
+            ]
+            .into_iter()
+            .collect();
+            let otel_keys: BTreeSet<&str> = [
+                "code.namespace",
+                "code.filepath",
+                "code.filename",
+                "code.lineno",
+                "expected",
+                "emitted",
+                "phase",
+            ]
+            .into_iter()
+            .collect();
+            let resource = isolated.exporter.resource.lock().expect("capture exporter resource");
+            assert!(
+                resource
+                    .as_ref()
+                    .expect("resource was passed to exporter")
+                    .iter()
+                    .next()
+                    .is_none()
+            );
+            for (phase, (entry, (log, scope))) in [0_u64, 1].into_iter().zip(json.iter().zip(&logs)) {
+                let object = entry.as_object().expect("diagnostic JSON must be an object");
+                assert!(object.keys().all(|key| json_keys.contains(key.as_str())), "unexpected JSON field");
+                assert_eq!(entry["target"].as_str(), Some(TARGET));
+                assert_eq!(entry["phase"].as_u64(), Some(phase));
+                assert_eq!(entry["expected"].as_u64(), Some(1024));
+                assert_eq!(entry["emitted"].as_u64(), Some(64));
+                assert!(!entry.to_string().contains(FAKE_URI));
+                assert!(!entry.to_string().contains(FAKE_AUTH));
+
+                assert_eq!(log.target().map(|target| target.as_ref()), Some(TARGET));
+                assert!(log.trace_context().is_none(), "ambient OTEL context must not reach the exporter");
+                assert_eq!(scope.name(), "");
+                assert!(scope.version().is_none());
+                assert!(scope.schema_url().is_none());
+                assert!(scope.attributes().next().is_none());
+                assert!(
+                    log.attributes_iter().all(|(key, _)| otel_keys.contains(key.as_str())),
+                    "unexpected OTEL field"
+                );
+                let phase = i64::try_from(phase).expect("fixture phase fits into i64");
+                assert!(
+                    log.attributes_iter()
+                        .any(|(key, value)| key.as_str() == "phase" && value == &AnyValue::Int(phase))
+                );
+                assert!(
+                    log.attributes_iter()
+                        .any(|(key, value)| key.as_str() == "expected" && value == &AnyValue::Int(1024))
+                );
+                assert!(
+                    log.attributes_iter()
+                        .any(|(key, value)| key.as_str() == "emitted" && value == &AnyValue::Int(64))
+                );
+                assert!(!format!("{log:?}").contains(FAKE_URI));
+                assert!(!format!("{log:?}").contains(FAKE_AUTH));
+            }
+        }
+    }
+}
+
+mod async_isolation {
+    use super::*;
+    use std::{
+        collections::HashSet,
+        future::pending,
+        panic::{AssertUnwindSafe, catch_unwind},
+        sync::atomic::{AtomicBool, AtomicUsize, Ordering},
+        time::Duration,
+    };
+    use tracing::{Instrument, instrument::WithSubscriber};
+    use tracing_subscriber::EnvFilter;
+
+    const PARTICIPANTS: usize = 4;
+    const WAIT: Duration = Duration::from_secs(30);
+
+    fn parent_span(dispatch: &Dispatch) -> tracing::Span {
+        dispatcher::with_default(dispatch, || {
+            tracing::info_span!("http-request", uri = FAKE_URI, authorization = FAKE_AUTH, request_id = "synthetic-id")
+        })
+    }
+
+    fn observe_once(dispatch: &Dispatch, parent: &Option<tracing::span::Id>, phase: u64) {
+        let _context = fake_ambient_context().attach();
+        assert_eq!(&tracing::Span::current().id(), parent);
+        emit_candidate(dispatch, phase);
+        assert_eq!(&tracing::Span::current().id(), parent);
+        assert_eq!(Context::current().span().span_context().trace_id(), TraceId::from(17));
+        assert!(Context::current().baggage().get("synthetic-secret").is_some());
+    }
+
+    fn assert_isolated_events(sinks: &Sinks, expected_phases: &[u64]) {
+        let json = sinks.json_records();
+        let logs = sinks.exported_records();
+        assert_eq!(json.len(), expected_phases.len(), "each event must reach the isolated JSON sink");
+        assert_eq!(logs.len(), expected_phases.len(), "each event must reach the isolated exporter");
+        let json_keys: BTreeSet<&str> = [
+            "timestamp",
+            "level",
+            "target",
+            "filename",
+            "line_number",
+            "threadName",
+            "threadId",
+            "message",
+            "expected",
+            "emitted",
+            "phase",
+        ]
+        .into_iter()
+        .collect();
+        let otel_keys: BTreeSet<&str> = [
+            "code.namespace",
+            "code.filepath",
+            "code.filename",
+            "code.lineno",
+            "expected",
+            "emitted",
+            "phase",
+        ]
+        .into_iter()
+        .collect();
+        let mut json_phases = Vec::new();
+        for entry in &json {
+            let object = entry.as_object().expect("diagnostic JSON must be an object");
+            assert!(object.keys().all(|key| json_keys.contains(key.as_str())), "unexpected JSON field");
+            assert_eq!(entry["target"].as_str(), Some(TARGET));
+            assert_eq!(entry["message"].as_str(), Some("diagnostic fixture"));
+            assert_eq!(entry["expected"].as_u64(), Some(1024));
+            assert_eq!(entry["emitted"].as_u64(), Some(64));
+            json_phases.push(entry["phase"].as_u64().expect("phase is an unsigned number"));
+            assert!(!entry.to_string().contains(FAKE_URI));
+            assert!(!entry.to_string().contains(FAKE_AUTH));
+        }
+        let mut otel_phases = Vec::new();
+        for (log, scope) in &logs {
+            assert_eq!(log.target().map(|target| target.as_ref()), Some(TARGET));
+            assert!(log.trace_context().is_none());
+            assert!(log.attributes_iter().all(|(key, _)| otel_keys.contains(key.as_str())));
+            assert!(matches!(log.body(), Some(AnyValue::String(body)) if body.as_str() == "diagnostic fixture"));
+            assert!(
+                log.attributes_iter()
+                    .any(|(key, value)| key.as_str() == "expected" && value == &AnyValue::Int(1024))
+            );
+            assert!(
+                log.attributes_iter()
+                    .any(|(key, value)| key.as_str() == "emitted" && value == &AnyValue::Int(64))
+            );
+            let phase = log
+                .attributes_iter()
+                .find_map(|(key, value)| match (key.as_str(), value) {
+                    ("phase", AnyValue::Int(phase)) => u64::try_from(*phase).ok(),
+                    _ => None,
+                })
+                .expect("phase is a nonnegative OTEL integer");
+            otel_phases.push(phase);
+            assert_eq!(scope.name(), "");
+            assert!(scope.version().is_none());
+            assert!(scope.schema_url().is_none());
+            assert!(scope.attributes().next().is_none());
+            assert!(!format!("{log:?}").contains(FAKE_URI));
+            assert!(!format!("{log:?}").contains(FAKE_AUTH));
+        }
+        json_phases.sort_unstable();
+        otel_phases.sort_unstable();
+        let mut expected_phases = expected_phases.to_vec();
+        expected_phases.sort_unstable();
+        assert_eq!(json_phases, expected_phases);
+        assert_eq!(otel_phases, expected_phases);
+        let resource = sinks.exporter.resource.lock().expect("capture exporter resource");
+        assert!(resource.as_ref().expect("exporter resource exists").iter().next().is_none());
+    }
+
+    #[test]
+    fn separate_threads_keep_their_contexts_across_async_suspension() {
+        let outer = Sinks::new(true, true);
+        let isolated = Sinks::new(true, true);
+        let barrier = Arc::new(tokio::sync::Barrier::new(PARTICIPANTS));
+        let thread_ids: HashSet<_> = std::thread::scope(|scope| {
+            let mut handles = Vec::new();
+            for _ in 0..PARTICIPANTS {
+                let outer_dispatch = outer.dispatch.clone();
+                let isolated_dispatch = isolated.dispatch.clone();
+                let barrier = Arc::clone(&barrier);
+                handles.push(scope.spawn(move || {
+                    let thread_id = std::thread::current().id();
+                    let span = parent_span(&outer_dispatch);
+                    let span_id = span.id();
+                    let runtime = tokio::runtime::Builder::new_current_thread()
+                        .enable_all()
+                        .build()
+                        .expect("build a test-only runtime");
+                    runtime.block_on(
+                        async move {
+                            for phase in [0_u64, 1] {
+                                tokio::time::timeout(WAIT, barrier.wait())
+                                    .await
+                                    .expect("all threads reach the barrier");
+                                observe_once(&isolated_dispatch, &span_id, phase);
+                                tokio::task::yield_now().await;
+                            }
+                        }
+                        .instrument(span)
+                        .with_subscriber(outer_dispatch),
+                    );
+                    thread_id
+                }));
+            }
+            handles
+                .into_iter()
+                .map(|handle| handle.join().expect("thread completes"))
+                .collect()
+        });
+        assert_eq!(thread_ids.len(), PARTICIPANTS, "the fixture must use distinct OS threads");
+        assert!(outer.json_records().is_empty());
+        assert!(outer.exported_records().is_empty());
+        assert_isolated_events(&isolated, &[0, 1, 0, 1, 0, 1, 0, 1]);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn shared_runtime_reenters_outer_context_after_each_await() {
+        let outer = Sinks::new(true, true);
+        let isolated = Sinks::new(true, true);
+        let barrier = Arc::new(tokio::sync::Barrier::new(PARTICIPANTS));
+        let mut tasks = Vec::new();
+        for _ in 0..PARTICIPANTS {
+            let outer_dispatch = outer.dispatch.clone();
+            let isolated_dispatch = isolated.dispatch.clone();
+            let barrier = Arc::clone(&barrier);
+            let span = parent_span(&outer_dispatch);
+            let span_id = span.id();
+            tasks.push(tokio::spawn(
+                async move {
+                    for phase in [0_u64, 1] {
+                        tokio::time::timeout(WAIT, barrier.wait())
+                            .await
+                            .expect("all tasks reach the barrier");
+                        observe_once(&isolated_dispatch, &span_id, phase);
+                        tokio::task::yield_now().await;
+                    }
+                }
+                .instrument(span)
+                .with_subscriber(outer_dispatch),
+            ));
+        }
+        for task in tasks {
+            tokio::time::timeout(WAIT, task)
+                .await
+                .expect("task finishes")
+                .expect("task succeeds");
+        }
+        assert!(outer.json_records().is_empty());
+        assert!(outer.exported_records().is_empty());
+        assert_isolated_events(&isolated, &[0, 1, 0, 1, 0, 1, 0, 1]);
+    }
+
+    #[test]
+    fn either_isolation_step_alone_leaves_ambient_context() {
+        let outer = Sinks::new(true, true);
+        let separate = Sinks::new(true, true);
+        dispatcher::with_default(&outer.dispatch, || {
+            let span = parent_span(&outer.dispatch);
+            let _span = span.enter();
+            let _context = fake_ambient_context().attach();
+            dispatcher::with_default(&separate.dispatch, || emit_root_event(0));
+            let _empty_context = Context::new().attach();
+            emit_root_event(1);
+        });
+
+        let separate_json = separate.json_records();
+        let separate_logs = separate.exported_records();
+        assert_eq!(separate_json.len(), 1);
+        assert_eq!(separate_logs.len(), 1);
+        assert!(separate_json[0].get("span").is_none());
+        assert_eq!(
+            separate_logs[0]
+                .0
+                .trace_context()
+                .expect("ambient trace still exists")
+                .trace_id,
+            TraceId::from(17)
+        );
+
+        let outer_json = outer.json_records();
+        let outer_logs = outer.exported_records();
+        assert_eq!(outer_json.len(), 1);
+        assert_eq!(outer_logs.len(), 1);
+        assert_eq!(outer_json[0]["span"]["uri"].as_str(), Some(FAKE_URI));
+        assert_eq!(outer_json[0]["span"]["authorization"].as_str(), Some(FAKE_AUTH));
+        assert!(outer_logs[0].0.trace_context().is_none());
+    }
+
+    fn filtered_sinks(filter: &str, sensitive_resource: bool) -> Sinks {
+        let json = MemoryWriter::default();
+        let exporter = MemoryExporter::default();
+        let resource = if sensitive_resource {
+            Resource::builder_empty()
+                .with_attributes([KeyValue::new("synthetic.resource", FAKE_AUTH)])
+                .build()
+        } else {
+            Resource::builder_empty().build()
+        };
+        let provider = SdkLoggerProvider::builder()
+            .with_resource(resource)
+            .with_simple_exporter(exporter.clone())
+            .build();
+        let bridge = OpenTelemetryTracingBridge::builder(&provider)
+            .with_tracing_span_attributes(TracingSpanAttributes::all())
+            .build();
+        let subscriber = tracing_subscriber::registry()
+            .with(EnvFilter::new(filter))
+            .with(build_json_log_layer(json.clone(), false, FmtSpan::NONE, true))
+            .with(bridge);
+        Sinks {
+            dispatch: Dispatch::new(subscriber),
+            json,
+            exporter,
+            _provider: provider,
+        }
+    }
+
+    #[test]
+    fn filters_and_resources_do_not_cross_dispatchers() {
+        let outer = filtered_sinks("info,rustfs_isolation_probe_test=off", true);
+        let isolated = filtered_sinks("off,rustfs_isolation_probe_test=info", false);
+        let disabled = filtered_sinks("off", false);
+        dispatcher::with_default(&outer.dispatch, || {
+            let span = parent_span(&outer.dispatch);
+            let _span = span.enter();
+            let parent_id = span.id();
+            observe_once(&isolated.dispatch, &parent_id, 0);
+            observe_once(&disabled.dispatch, &parent_id, 1);
+            tracing::info!(target: "isolation_outer_control", value = 1_u64, "outer control");
+        });
+        let outer_json = outer.json_records();
+        let outer_logs = outer.exported_records();
+        assert_eq!(outer_json.len(), 1);
+        assert_eq!(outer_logs.len(), 1);
+        assert_eq!(outer_json[0]["target"].as_str(), Some("isolation_outer_control"));
+        assert_eq!(outer_logs[0].0.target().map(|target| target.as_ref()), Some("isolation_outer_control"));
+        assert!(
+            outer
+                .exporter
+                .resource
+                .lock()
+                .expect("capture outer resource")
+                .as_ref()
+                .expect("outer resource exists")
+                .get(&opentelemetry::Key::new("synthetic.resource"))
+                .is_some()
+        );
+        assert_isolated_events(&isolated, &[0]);
+        assert!(disabled.json_records().is_empty());
+        assert!(disabled.exported_records().is_empty());
+    }
+
+    struct CancellationProbe {
+        outer: Dispatch,
+        isolated: Dispatch,
+        drops: Arc<AtomicUsize>,
+        restored: Arc<AtomicBool>,
+    }
+
+    impl Drop for CancellationProbe {
+        fn drop(&mut self) {
+            dispatcher::with_default(&self.outer, || {
+                let span = parent_span(&self.outer);
+                let _span = span.enter();
+                let _context = fake_ambient_context().attach();
+                emit_candidate(&self.isolated, 1);
+                let restored = tracing::Span::current().id() == span.id()
+                    && Context::current().span().span_context().trace_id() == TraceId::from(17)
+                    && Context::current().baggage().get("synthetic-secret").is_some();
+                self.restored.store(restored, Ordering::SeqCst);
+            });
+            self.drops.fetch_add(1, Ordering::SeqCst);
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn cancellation_drop_emits_in_isolation_and_restores_context() {
+        let outer = Sinks::new(true, true);
+        let isolated = Sinks::new(true, true);
+        let drops = Arc::new(AtomicUsize::new(0));
+        let restored = Arc::new(AtomicBool::new(false));
+        let probe = CancellationProbe {
+            outer: outer.dispatch.clone(),
+            isolated: isolated.dispatch.clone(),
+            drops: Arc::clone(&drops),
+            restored: Arc::clone(&restored),
+        };
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let task = tokio::spawn(async move {
+            let _probe = probe;
+            started_tx.send(()).expect("signal that the cancellation fixture is active");
+            pending::<()>().await;
+        });
+        tokio::time::timeout(WAIT, started_rx)
+            .await
+            .expect("task starts")
+            .expect("start signal");
+        task.abort();
+        let error = tokio::time::timeout(WAIT, task)
+            .await
+            .expect("cancelled task finishes")
+            .expect_err("the task must be cancelled");
+        assert!(error.is_cancelled());
+        assert_eq!(drops.load(Ordering::SeqCst), 1);
+        assert!(restored.load(Ordering::SeqCst));
+        assert!(outer.json_records().is_empty());
+        assert!(outer.exported_records().is_empty());
+        assert_isolated_events(&isolated, &[1]);
+    }
+
+    #[derive(Clone)]
+    struct PanicWriter;
+
+    impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for PanicWriter {
+        type Writer = Self;
+
+        fn make_writer(&'a self) -> Self::Writer {
+            self.clone()
+        }
+    }
+
+    impl Write for PanicWriter {
+        fn write(&mut self, _: &[u8]) -> io::Result<usize> {
+            panic!("synthetic writer panic");
+        }
+
+        fn flush(&mut self) -> io::Result<()> {
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn panic_during_sink_write_restores_the_outer_context() {
+        let outer = Sinks::new(true, true);
+        let panic_dispatch =
+            Dispatch::new(tracing_subscriber::registry().with(build_json_log_layer(PanicWriter, false, FmtSpan::NONE, true)));
+        dispatcher::with_default(&outer.dispatch, || {
+            let span = parent_span(&outer.dispatch);
+            let _span = span.enter();
+            let _context = fake_ambient_context().attach();
+            let panic = catch_unwind(AssertUnwindSafe(|| emit_candidate(&panic_dispatch, 0)));
+            assert!(panic.is_err(), "the fixture must panic while writing the isolated event");
+            assert_eq!(tracing::Span::current().id(), span.id());
+            assert_eq!(Context::current().span().span_context().trace_id(), TraceId::from(17));
+            assert!(Context::current().baggage().get("synthetic-secret").is_some());
+            tracing::info!(target: "isolation_outer_control", value = 1_u64, "outer control");
+        });
+        let json = outer.json_records();
+        let logs = outer.exported_records();
+        assert_eq!(json.len(), 1);
+        assert_eq!(logs.len(), 1);
+        assert_eq!(json[0]["target"].as_str(), Some("isolation_outer_control"));
+        assert_eq!(logs[0].0.target().map(|target| target.as_ref()), Some("isolation_outer_control"));
+    }
+}

crates/obs/src/telemetry/mod.rs

@@ -43,6 +43,8 @@
 pub mod dial9;
 mod filter;
 mod guard;
+#[cfg(test)]
+mod isolation_probe_tests;
 mod local;
 mod otel;
 mod recorder;