Run scoped scanner Linux diagnostics

.github/workflows/windows-filesystem.yml

@@ -1,224 +1,65 @@
-# 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
-
+name: Scanner Linux Hosted 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
-
+  group: scanner-linux-hosted-${{ github.ref }}
+  cancel-in-progress: false
 env:
-  CARGO_TERM_COLOR: always
-  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+  CARGO_TERM_COLOR: never
   RUST_BACKTRACE: 1
-
+  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: 'true'
+  NO_PROXY: 127.0.0.1,localhost
 jobs:
-  rename-safety:
-    name: Rename Safety
-    runs-on: windows-latest
-    timeout-minutes: 60
-    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: 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
+  scoped-tests:
+    runs-on: ubuntu-latest
     timeout-minutes: 90
     steps:
-      - name: Checkout repository
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+      - name: Checkout candidate
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a
         with:
           persist-credentials: false
-
-      - name: Setup Rust environment
+          fetch-depth: 2
+      - name: Setup Rust
         uses: ./.github/actions/setup
         with:
           rust-version: stable
-          cache-shared-key: build-x86_64-pc-windows-msvc
+          cache-shared-key: ci-e2e-distributed-hosted
           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
+      - name: Check candidate and resources
+        shell: bash
         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
+          set -euo pipefail
+          test "$(git rev-parse HEAD^)" = 03eef1ea79a782e15480a00ba87ed2e256a84f9b
+          test "$(git diff --name-only HEAD^ HEAD)" = .github/workflows/windows-filesystem.yml
+          df -h .
+          free -h
+          du -sh target "$HOME/.cargo" 2>/dev/null || true
+      - name: Run only the named diagnostic unit tests
+        shell: bash
         env:
-          RUSTFS_ACCESS_KEY: rustfsadmin
-          RUSTFS_SECRET_KEY: rustfsadmin
+          CARGO_BUILD_JOBS: '1'
+          CARGO_INCREMENTAL: '0'
         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
+          run_one() {
+            local package="$1"
+            local test_name="$2"
+            cargo test --locked -p "$package" --lib -- --exact "$test_name" --nocapture 2>&1 | tee scanner-proof-test.log
+            grep -F 'test result: ok. 1 passed; 0 failed;' scanner-proof-test.log
           }
+          run_one rustfs-scanner scanner::tests::cycle_budget_publication::publication_lease_observations_keep_cycle_epoch_purpose_and_outcome
+          run_one rustfs-scanner scanner::tests::cycle_budget_publication::publication_lease_retry_logs_identify_interleaved_epochs
+          run_one rustfs-scanner scanner::tests::cycle_budget_publication::publication_lease_events_include_context_in_otlp_without_span_export
+          run_one rustfs-scanner scanner::tests::cycle_budget_publication::dropped_lease_wait_does_not_report_a_finished_outcome
+          run_one rustfs-scanner scanner::tests::cycle_budget_publication::publication_lease_retries_local_channel_deadline_and_stops_on_cancel
+          run_one rustfs-ecstore cluster::rpc::peer_rest_client::tests::scanner_publication_lease_validation_errors_are_typed_without_changing_display
+          run_one rustfs-ecstore cluster::rpc::peer_rest_client::tests::scanner_publication_lease_error_class_keeps_remote_reports_separate_from_local_errors
+          run_one rustfs-ecstore services::notification_sys::tests::scanner_lease_acquisition_error_identifies_peer_and_preserves_retry_suffix
+          run_one rustfs-ecstore services::notification_sys::tests::scanner_lease_failure_info_uses_the_actual_target_and_rejects_unbounded_peers
+          run_one rustfs-ecstore services::notification_sys::tests::scanner_lease_failure_info_accepts_real_grid_host_urls
+          run_one rustfs-ecstore services::notification_sys::tests::scanner_lease_failure_info_preserves_typed_connection_failures_without_logging_details
+          run_one rustfs-ecstore services::notification_sys::tests::scanner_lease_channel_deadline_preserves_public_error_and_diagnostic

Cargo.lock

@@ -10642,6 +10642,9 @@ dependencies = [
  "hotpath",
  "http 1.5.0",
  "metrics",
+ "opentelemetry",
+ "opentelemetry-appender-tracing",
+ "opentelemetry_sdk",
  "rand 0.10.3",
  "rmp-serde",
  "rustfs-common",

crates/common/src/globals.rs

@@ -261,6 +261,20 @@ pub async fn cached_connection(addr: &str) -> Option<Channel> {
     GLOBAL_CONN_MAP.read().await.get(addr).cloned()
 }
 
+/// Holds the connection cache write lock for an isolated test process.
+#[cfg(any(test, feature = "test-util"))]
+pub struct ConnectionCacheReadBlockerForTest {
+    _guard: tokio::sync::RwLockWriteGuard<'static, HashMap<String, Channel>>,
+}
+
+/// Block cache reads until the returned test guard is dropped.
+#[cfg(any(test, feature = "test-util"))]
+pub async fn block_connection_cache_reads_for_test() -> ConnectionCacheReadBlockerForTest {
+    ConnectionCacheReadBlockerForTest {
+        _guard: GLOBAL_CONN_MAP.write().await,
+    }
+}
+
 /// Cache a gRPC connection for the given address.
 pub async fn cache_connection(addr: String, channel: Channel) {
     GLOBAL_CONN_MAP.write().await.insert(addr, channel);

crates/ecstore/Cargo.toml

@@ -245,6 +245,7 @@ windows-sys = { workspace = true, features = ["Win32_System_Ioctl"] }
 
 [dev-dependencies]
 aws-smithy-async.workspace = true
+rustfs-common = { workspace = true, features = ["test-util"] }
 rustfs-s3-ops.workspace = true
 tokio = { workspace = true, features = ["rt-multi-thread", "macros", "test-util", "fs"] }
 criterion = { workspace = true, features = ["html_reports"] }

crates/ecstore/src/api/mod.rs

@@ -527,15 +527,15 @@ pub mod notification {
     pub use crate::services::notification_sys::{
         ClusterTierDailyStats, CrossPoolFenceFleetProofToken, IlmRecoveryExportFleetProofToken,
         LegacyTransitionStateReconcileFleetProofToken, MrfLifecycleWriterFleetProofToken, NotificationPeerErr, NotificationSys,
-        ScannerPublicationLeaseGrant, acquire_cross_pool_fence_fleet_proof, acquire_ilm_recovery_export_fleet_proof,
-        acquire_legacy_transition_state_reconcile_fleet_proof, acquire_mrf_lifecycle_writer_fleet_proof,
-        configured_fleet_topology_fingerprint, cross_pool_fence_fleet_proof_matches, get_global_notification_sys,
-        ilm_recovery_export_fleet_proof_matches, ilm_recovery_export_local_process_epoch,
+        ScannerPublicationLeaseFailureInfo, ScannerPublicationLeaseGrant, acquire_cross_pool_fence_fleet_proof,
+        acquire_ilm_recovery_export_fleet_proof, acquire_legacy_transition_state_reconcile_fleet_proof,
+        acquire_mrf_lifecycle_writer_fleet_proof, configured_fleet_topology_fingerprint, cross_pool_fence_fleet_proof_matches,
+        get_global_notification_sys, ilm_recovery_export_fleet_proof_matches, ilm_recovery_export_local_process_epoch,
         ilm_recovery_export_member_epochs_sha256, ilm_recovery_export_topology_generation,
         legacy_transition_state_reconcile_fleet_proof_matches, mrf_lifecycle_reader_local_process_epoch,
         mrf_lifecycle_writer_fleet_proof_current, mrf_lifecycle_writer_fleet_proof_matches, new_global_notification_sys,
-        scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe,
-        with_mrf_lifecycle_writer_fleet_proof,
+        scanner_peer_transport_error_message_is_retryable, scanner_publication_lease_failure_info,
+        start_remote_version_state_fleet_probe, with_mrf_lifecycle_writer_fleet_proof,
     };
     #[cfg(any(test, feature = "test-util"))]
     pub use crate::services::notification_sys::{

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -668,6 +668,136 @@ impl ScannerPublicationLease {
     }
 }
 
+struct ScannerPublicationLeaseLocalError {
+    reason: &'static str,
+    detail: String,
+}
+
+impl std::fmt::Debug for ScannerPublicationLeaseLocalError {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("ScannerPublicationLeaseLocalError")
+            .field("reason", &self.reason)
+            .finish()
+    }
+}
+
+impl std::fmt::Display for ScannerPublicationLeaseLocalError {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str(&self.detail)
+    }
+}
+
+impl std::error::Error for ScannerPublicationLeaseLocalError {}
+
+struct ScannerPublicationLeaseRemoteRejection(String);
+
+impl std::fmt::Debug for ScannerPublicationLeaseRemoteRejection {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str("ScannerPublicationLeaseRemoteRejection")
+    }
+}
+
+impl std::fmt::Display for ScannerPublicationLeaseRemoteRejection {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str(&self.0)
+    }
+}
+
+impl std::error::Error for ScannerPublicationLeaseRemoteRejection {}
+
+pub(crate) fn scanner_publication_lease_remote_rejection(detail: String) -> Error {
+    Error::other(ScannerPublicationLeaseRemoteRejection(detail))
+}
+
+fn scanner_publication_lease_local_error(reason: &'static str, detail: &'static str) -> Error {
+    Error::other(ScannerPublicationLeaseLocalError {
+        reason,
+        detail: detail.to_owned(),
+    })
+}
+
+fn scanner_publication_lease_local_io_error(reason: &'static str, error: std::io::Error) -> Error {
+    Error::Io(std::io::Error::new(
+        error.kind(),
+        ScannerPublicationLeaseLocalError {
+            reason,
+            detail: error.to_string(),
+        },
+    ))
+}
+
+#[derive(Debug)]
+pub(crate) struct ScannerPublicationLeaseErrorClass {
+    pub reason: Option<&'static str>,
+    pub reported_reason: Option<&'static str>,
+}
+
+pub(crate) fn scanner_publication_lease_error_class(error: &Error) -> ScannerPublicationLeaseErrorClass {
+    let mut result = ScannerPublicationLeaseErrorClass {
+        reason: None,
+        reported_reason: None,
+    };
+    match error {
+        Error::RemoteClientUnavailable(_) => result.reason = Some("client_unavailable"),
+        Error::RemoteNotInitialized => result.reason = Some("peer_not_initialized"),
+        Error::Timeout => result.reason = Some("timeout"),
+        Error::Io(error) => {
+            if let Some(local) = error
+                .get_ref()
+                .and_then(|source| source.downcast_ref::<ScannerPublicationLeaseLocalError>())
+            {
+                result.reason = Some(local.reason);
+            } else if let Some(remote) = error
+                .get_ref()
+                .and_then(|source| source.downcast_ref::<ScannerPublicationLeaseRemoteRejection>())
+            {
+                // These are the exact Display strings currently produced by
+                // the peer's lease acquisition path, not arbitrary suffixes.
+                result.reported_reason = match remote.0.as_str() {
+                    "Io error: scanner publication lease capacity is exhausted" => Some("capacity_exhausted"),
+                    "Io error: scanner publication lease generation is stale" => Some("generation_stale"),
+                    "Io error: scanner publication lease is blocked by data movement" => Some("blocked_by_data_movement"),
+                    _ => None,
+                };
+            } else if let Some(status) = embedded_tonic_status(error) {
+                result.reason = Some(match status.code() {
+                    tonic::Code::Ok => "rpc_ok",
+                    tonic::Code::Cancelled => "rpc_cancelled",
+                    tonic::Code::Unknown => "rpc_unknown",
+                    tonic::Code::InvalidArgument => "rpc_invalid_argument",
+                    tonic::Code::DeadlineExceeded => "rpc_deadline_exceeded",
+                    tonic::Code::NotFound => "rpc_not_found",
+                    tonic::Code::AlreadyExists => "rpc_already_exists",
+                    tonic::Code::PermissionDenied => "rpc_permission_denied",
+                    tonic::Code::ResourceExhausted => "rpc_resource_exhausted",
+                    tonic::Code::FailedPrecondition => "rpc_failed_precondition",
+                    tonic::Code::Aborted => "rpc_aborted",
+                    tonic::Code::OutOfRange => "rpc_out_of_range",
+                    tonic::Code::Unimplemented => "rpc_unimplemented",
+                    tonic::Code::Internal => "rpc_internal",
+                    tonic::Code::Unavailable => "rpc_unavailable",
+                    tonic::Code::DataLoss => "rpc_data_loss",
+                    tonic::Code::Unauthenticated => "rpc_unauthenticated",
+                });
+            } else {
+                result.reason = match error.kind() {
+                    std::io::ErrorKind::TimedOut => Some("io_timeout"),
+                    std::io::ErrorKind::ConnectionRefused => Some("io_connection_refused"),
+                    std::io::ErrorKind::ConnectionReset => Some("io_connection_reset"),
+                    std::io::ErrorKind::ConnectionAborted => Some("io_connection_aborted"),
+                    std::io::ErrorKind::BrokenPipe => Some("io_broken_pipe"),
+                    std::io::ErrorKind::UnexpectedEof => Some("io_unexpected_eof"),
+                    std::io::ErrorKind::NotConnected => Some("io_not_connected"),
+                    _ => None,
+                };
+            }
+        }
+        _ => {}
+    }
+    result
+}
+
 fn validate_scanner_publication_lease_response_fields(
     response: &ScannerPublicationLeaseResponse,
     expected_session_id: &str,
@@ -675,7 +805,7 @@ fn validate_scanner_publication_lease_response_fields(
     expected_observational_only: bool,
 ) -> Result<(Uuid, String)> {
     if !response.success {
-        return Err(Error::other(
+        return Err(scanner_publication_lease_remote_rejection(
             response
                 .error
                 .as_ref()
@@ -684,24 +814,39 @@ fn validate_scanner_publication_lease_response_fields(
         ));
     }
     if response.movement_generation != expected_generation {
-        return Err(Error::other("peer returned a different scanner publication lease generation"));
+        return Err(scanner_publication_lease_local_error(
+            "generation_mismatch",
+            "peer returned a different scanner publication lease generation",
+        ));
     }
     if response.session_id != expected_session_id {
-        return Err(Error::other("peer returned a different scanner publication lease session"));
+        return Err(scanner_publication_lease_local_error(
+            "session_mismatch",
+            "peer returned a different scanner publication lease session",
+        ));
     }
     if response.observational_only != expected_observational_only {
-        return Err(Error::other("peer returned a different scanner publication lease purpose"));
+        return Err(scanner_publication_lease_local_error(
+            "purpose_mismatch",
+            "peer returned a different scanner publication lease purpose",
+        ));
     }
     let owner_id = Uuid::parse_str(&response.owner_id)
         .ok()
         .filter(|owner_id| !owner_id.is_nil())
         .map(|owner_id| owner_id.to_string())
-        .ok_or_else(|| Error::other("peer returned an invalid scanner publication lease owner"))?;
+        .ok_or_else(|| {
+            scanner_publication_lease_local_error("invalid_owner", "peer returned an invalid scanner publication lease owner")
+        })?;
     if response.lease_ttl_ms != crate::store::SCANNER_PUBLICATION_LEASE_TTL_MS {
-        return Err(Error::other("peer returned an unsupported scanner publication lease TTL"));
+        return Err(scanner_publication_lease_local_error(
+            "unsupported_ttl",
+            "peer returned an unsupported scanner publication lease TTL",
+        ));
     }
-    let token = Uuid::from_slice(response.token.as_ref())
-        .map_err(|_| Error::other("peer returned an invalid scanner publication lease token"))?;
+    let token = Uuid::from_slice(response.token.as_ref()).map_err(|_| {
+        scanner_publication_lease_local_error("invalid_token", "peer returned an invalid scanner publication lease token")
+    })?;
     Ok((token, owner_id))
 }
 
@@ -712,16 +857,27 @@ fn scanner_publication_lease_deadline(
 ) -> Result<std::time::Instant> {
     let lease_window = Duration::from_millis(lease_ttl_ms)
         .checked_sub(SCANNER_PUBLICATION_LEASE_SAFETY_MARGIN)
-        .ok_or_else(|| Error::other("scanner publication lease TTL is shorter than its safety margin"))?;
-    let elapsed = response_received
-        .checked_duration_since(request_started)
-        .ok_or_else(|| Error::other("scanner publication lease response clock moved backwards"))?;
+        .ok_or_else(|| {
+            scanner_publication_lease_local_error(
+                "ttl_below_safety_margin",
+                "scanner publication lease TTL is shorter than its safety margin",
+            )
+        })?;
+    let elapsed = response_received.checked_duration_since(request_started).ok_or_else(|| {
+        scanner_publication_lease_local_error(
+            "response_clock_backwards",
+            "scanner publication lease response clock moved backwards",
+        )
+    })?;
     if elapsed >= lease_window {
-        return Err(Error::other("scanner publication lease response arrived after its safety window"));
+        return Err(scanner_publication_lease_local_error(
+            "response_after_safety_window",
+            "scanner publication lease response arrived after its safety window",
+        ));
     }
-    request_started
-        .checked_add(lease_window)
-        .ok_or_else(|| Error::other("scanner publication lease deadline overflowed"))
+    request_started.checked_add(lease_window).ok_or_else(|| {
+        scanner_publication_lease_local_error("deadline_overflow", "scanner publication lease deadline overflowed")
+    })
 }
 
 /// A peer's readiness view of one drive served by another host.
@@ -876,6 +1032,11 @@ fn tier_mutation_failed_response_error(version: u32, failure_class: i32, error_i
     Error::other(message)
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum ScannerLeaseConnectionFailure {
+    ChannelAcquisitionTimedOut,
+}
+
 impl PeerRestClient {
     fn recovery_monitor_span(grid_host: &str) -> tracing::Span {
         tracing::info_span!(
@@ -1003,19 +1164,37 @@ impl PeerRestClient {
     }
 
     pub async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>> {
+        self.get_client_with_connection_failure().await.map_err(|(error, _)| error)
+    }
+
+    async fn get_client_with_connection_failure(
+        &self,
+    ) -> std::result::Result<
+        NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>,
+        (Error, Option<ScannerLeaseConnectionFailure>),
+    > {
         if self.offline.load(Ordering::Acquire) {
             self.mark_offline_and_spawn_recovery();
-            return Err(Error::RemoteClientUnavailable(format!("peer {} is temporarily offline", self.grid_host)));
+            return Err((
+                Error::RemoteClientUnavailable(format!("peer {} is temporarily offline", self.grid_host)),
+                None,
+            ));
         }
 
         node_service_time_out_client(&self.grid_host, TonicInterceptor::Signature(gen_tonic_signature_interceptor()))
             .await
             .map_err(|err| {
+                // Channel acquisition reports its deadline as a top-level
+                // typed I/O timeout. Do not infer kinds from dial error text.
+                let failure = err
+                    .downcast_ref::<std::io::Error>()
+                    .filter(|error| error.kind() == std::io::ErrorKind::TimedOut)
+                    .map(|_| ScannerLeaseConnectionFailure::ChannelAcquisitionTimedOut);
                 let storage_err = Error::RemoteClientUnavailable(format!("can not get client, err: {err}"));
                 if Self::is_network_like_error(&storage_err) {
                     self.mark_offline_and_spawn_recovery();
                 }
-                storage_err
+                (storage_err, failure)
             })
     }
 
@@ -2508,54 +2687,91 @@ impl PeerRestClient {
         expected_generation: u64,
         observational_only: bool,
     ) -> Result<ScannerPublicationLease> {
-        let request_started = std::time::Instant::now();
-        self.finalize_result(
-            async {
-                let challenge = Uuid::new_v4();
-                let mut client = self
-                    .get_client()
-                    .await?
-                    .max_decoding_message_size(SCANNER_ACTIVITY_MAX_MESSAGE_SIZE)
-                    .max_encoding_message_size(SCANNER_ACTIVITY_MAX_MESSAGE_SIZE);
-                let mut request = Request::new(ScannerPublicationLeaseRequest {
-                    challenge: challenge.as_bytes().to_vec().into(),
-                    expected_movement_generation: expected_generation,
-                    ttl_ms: crate::store::SCANNER_PUBLICATION_LEASE_TTL_MS,
-                    expected_session_id: expected_session_id.to_string(),
-                    token: Bytes::new(),
-                    observational_only,
-                });
-                let canonical = rustfs_protos::canonical_scanner_publication_lease_request_body(request.get_ref())
-                    .map_err(|_| Error::other("scanner publication lease request is too large to authenticate"))?;
-                set_tonic_canonical_body_digest(&mut request, &canonical)?;
-                let response = client.acquire_scanner_publication_lease(request).await?.into_inner();
-                let response_body =
-                    rustfs_protos::canonical_scanner_publication_lease_response_body(challenge.as_bytes(), &response)
-                        .map_err(|_| Error::other("scanner publication lease response is too large to authenticate"))?;
-                verify_tonic_rpc_response_proof(&response_body, &response.response_proof)
-                    .map_err(|_| Error::other("peer returned an invalid scanner publication lease proof"))?;
-                let (token, owner_id) = validate_scanner_publication_lease_response_fields(
-                    &response,
-                    expected_session_id,
-                    expected_generation,
-                    observational_only,
-                )?;
-                Ok(ScannerPublicationLease {
-                    token,
-                    movement_generation: response.movement_generation,
-                    observational_only,
-                    owner_id,
-                    session_id: response.session_id,
-                    expires_at: scanner_publication_lease_deadline(
-                        request_started,
-                        std::time::Instant::now(),
-                        response.lease_ttl_ms,
-                    )?,
-                })
-            }
-            .await,
+        self.acquire_scanner_publication_lease_with_connection_failure(
+            expected_session_id,
+            expected_generation,
+            observational_only,
         )
         .await
+        .map_err(|(error, _)| error)
+    }
+
+    pub(crate) async fn acquire_scanner_publication_lease_with_connection_failure(
+        &self,
+        expected_session_id: &str,
+        expected_generation: u64,
+        observational_only: bool,
+    ) -> std::result::Result<ScannerPublicationLease, (Error, Option<ScannerLeaseConnectionFailure>)> {
+        let request_started = std::time::Instant::now();
+        let mut connection_failure = None;
+        let result = self
+            .finalize_result(
+                async {
+                    let challenge = Uuid::new_v4();
+                    let mut client = self
+                        .get_client_with_connection_failure()
+                        .await
+                        .map_err(|(error, failure)| {
+                            connection_failure = failure;
+                            error
+                        })?
+                        .max_decoding_message_size(SCANNER_ACTIVITY_MAX_MESSAGE_SIZE)
+                        .max_encoding_message_size(SCANNER_ACTIVITY_MAX_MESSAGE_SIZE);
+                    let mut request = Request::new(ScannerPublicationLeaseRequest {
+                        challenge: challenge.as_bytes().to_vec().into(),
+                        expected_movement_generation: expected_generation,
+                        ttl_ms: crate::store::SCANNER_PUBLICATION_LEASE_TTL_MS,
+                        expected_session_id: expected_session_id.to_string(),
+                        token: Bytes::new(),
+                        observational_only,
+                    });
+                    let canonical =
+                        rustfs_protos::canonical_scanner_publication_lease_request_body(request.get_ref()).map_err(|_| {
+                            scanner_publication_lease_local_error(
+                                "request_too_large",
+                                "scanner publication lease request is too large to authenticate",
+                            )
+                        })?;
+                    set_tonic_canonical_body_digest(&mut request, &canonical)
+                        .map_err(|error| scanner_publication_lease_local_io_error("request_digest_failed", error))?;
+                    let response = client.acquire_scanner_publication_lease(request).await?.into_inner();
+                    let response_body =
+                        rustfs_protos::canonical_scanner_publication_lease_response_body(challenge.as_bytes(), &response)
+                            .map_err(|_| {
+                                scanner_publication_lease_local_error(
+                                    "response_too_large",
+                                    "scanner publication lease response is too large to authenticate",
+                                )
+                            })?;
+                    verify_tonic_rpc_response_proof(&response_body, &response.response_proof).map_err(|_| {
+                        scanner_publication_lease_local_error(
+                            "invalid_proof",
+                            "peer returned an invalid scanner publication lease proof",
+                        )
+                    })?;
+                    let (token, owner_id) = validate_scanner_publication_lease_response_fields(
+                        &response,
+                        expected_session_id,
+                        expected_generation,
+                        observational_only,
+                    )?;
+                    Ok(ScannerPublicationLease {
+                        token,
+                        movement_generation: response.movement_generation,
+                        observational_only,
+                        owner_id,
+                        session_id: response.session_id,
+                        expires_at: scanner_publication_lease_deadline(
+                            request_started,
+                            std::time::Instant::now(),
+                            response.lease_ttl_ms,
+                        )?,
+                    })
+                }
+                .await,
+            )
+            .await;
+        result.map_err(|error| (error, connection_failure))
     }
 
     /// Revalidate the exact token immediately before the coordinator commits
@@ -3146,6 +3362,162 @@ mod tests {
         let error = scanner_publication_lease_deadline(started, started + Duration::from_secs(55), 60_000)
             .expect_err("a response arriving at the safety boundary must fail closed");
         assert!(error.to_string().contains("after its safety window"));
+        assert_eq!(scanner_publication_lease_error_class(&error).reason, Some("response_after_safety_window"));
+        assert!(
+            error
+                .to_string()
+                .ends_with("scanner publication lease response arrived after its safety window")
+        );
+    }
+
+    #[test]
+    fn scanner_publication_lease_validation_errors_are_typed_without_changing_display() {
+        let valid = ScannerPublicationLeaseResponse {
+            success: true,
+            token: Uuid::new_v4().as_bytes().to_vec().into(),
+            movement_generation: 7,
+            lease_ttl_ms: crate::store::SCANNER_PUBLICATION_LEASE_TTL_MS,
+            error: None,
+            response_proof: Bytes::new(),
+            owner_id: Uuid::new_v4().to_string(),
+            session_id: "session-a".to_owned(),
+            observational_only: false,
+        };
+        let cases = [
+            (
+                ScannerPublicationLeaseResponse {
+                    movement_generation: 6,
+                    ..valid.clone()
+                },
+                "generation_mismatch",
+                "peer returned a different scanner publication lease generation",
+            ),
+            (
+                ScannerPublicationLeaseResponse {
+                    session_id: "other".to_owned(),
+                    ..valid.clone()
+                },
+                "session_mismatch",
+                "peer returned a different scanner publication lease session",
+            ),
+            (
+                ScannerPublicationLeaseResponse {
+                    observational_only: true,
+                    ..valid.clone()
+                },
+                "purpose_mismatch",
+                "peer returned a different scanner publication lease purpose",
+            ),
+            (
+                ScannerPublicationLeaseResponse {
+                    owner_id: Uuid::nil().to_string(),
+                    ..valid.clone()
+                },
+                "invalid_owner",
+                "peer returned an invalid scanner publication lease owner",
+            ),
+            (
+                ScannerPublicationLeaseResponse {
+                    lease_ttl_ms: 0,
+                    ..valid.clone()
+                },
+                "unsupported_ttl",
+                "peer returned an unsupported scanner publication lease TTL",
+            ),
+            (
+                ScannerPublicationLeaseResponse {
+                    token: Bytes::from_static(b"short"),
+                    ..valid.clone()
+                },
+                "invalid_token",
+                "peer returned an invalid scanner publication lease token",
+            ),
+        ];
+        for (response, reason, message) in cases {
+            let error = validate_scanner_publication_lease_response_fields(&response, "session-a", 7, false)
+                .expect_err("invalid lease response must be rejected");
+            let class = scanner_publication_lease_error_class(&error);
+            assert_eq!(class.reason, Some(reason));
+            assert_eq!(class.reported_reason, None);
+            assert_eq!(error.to_string(), format!("Io error: {message}"));
+            assert!(matches!(error, Error::Io(ref io) if io.kind() == std::io::ErrorKind::Other));
+        }
+
+        for (reason, message) in [
+            ("request_too_large", "scanner publication lease request is too large to authenticate"),
+            ("response_too_large", "scanner publication lease response is too large to authenticate"),
+            ("invalid_proof", "peer returned an invalid scanner publication lease proof"),
+            ("deadline_overflow", "scanner publication lease deadline overflowed"),
+        ] {
+            let error = scanner_publication_lease_local_error(reason, message);
+            assert_eq!(scanner_publication_lease_error_class(&error).reason, Some(reason));
+            assert_eq!(error.to_string(), format!("Io error: {message}"));
+            assert!(matches!(error, Error::Io(ref io) if io.kind() == std::io::ErrorKind::Other));
+        }
+
+        let original = std::io::Error::new(std::io::ErrorKind::InvalidInput, "digest rejected");
+        let expected = format!("Io error: {original}");
+        let wrapped = scanner_publication_lease_local_io_error("request_digest_failed", original);
+        assert_eq!(scanner_publication_lease_error_class(&wrapped).reason, Some("request_digest_failed"));
+        assert_eq!(wrapped.to_string(), expected);
+        assert!(matches!(wrapped, Error::Io(ref io) if io.kind() == std::io::ErrorKind::InvalidInput));
+    }
+
+    #[test]
+    fn scanner_publication_lease_error_class_keeps_remote_reports_separate_from_local_errors() {
+        for (remote_message, reported) in [
+            ("Io error: scanner publication lease capacity is exhausted", Some("capacity_exhausted")),
+            ("Io error: scanner publication lease generation is stale", Some("generation_stale")),
+            (
+                "Io error: scanner publication lease is blocked by data movement",
+                Some("blocked_by_data_movement"),
+            ),
+            ("remote-secret Io error: scanner publication lease generation is stale", None),
+            ("not Io error: scanner publication lease generation is stale", None),
+            ("scanner publication lease generation is stale", None),
+            ("Io error: scanner publication lease generation is stale ", None),
+            ("Io error: scanner publication lease is blocked by pending namespace commit", None),
+            ("Io error: scanner publication lease namespace changed during acquisition", None),
+        ] {
+            let rejected = ScannerPublicationLeaseResponse {
+                success: false,
+                error: Some(rustfs_protos::proto_gen::node_service::Error {
+                    code: 0,
+                    error_info: remote_message.to_owned(),
+                }),
+                ..Default::default()
+            };
+            let error = validate_scanner_publication_lease_response_fields(&rejected, "session", 0, false)
+                .expect_err("peer rejection must be returned");
+            let class = scanner_publication_lease_error_class(&error);
+            assert_eq!(class.reason, None);
+            assert_eq!(class.reported_reason, reported);
+            assert_eq!(error.to_string(), format!("Io error: {remote_message}"));
+            assert!(!format!("{error:?}").contains("remote-secret"));
+        }
+
+        let forged = Error::other("peer returned a different scanner publication lease generation");
+        let class = scanner_publication_lease_error_class(&forged);
+        assert_eq!(class.reason, None, "untyped error text must not be promoted to a local diagnosis");
+        assert_eq!(class.reported_reason, None, "untyped text must not be attributed to a peer response");
+
+        for (error, expected) in [
+            (Error::RemoteClientUnavailable("remote-secret".to_owned()), "client_unavailable"),
+            (Error::RemoteClientUnavailable("pretend timed out".to_owned()), "client_unavailable"),
+            (Error::Timeout, "timeout"),
+            (tonic::Status::unavailable("remote-secret").into(), "rpc_unavailable"),
+            (tonic::Status::deadline_exceeded("remote-secret").into(), "rpc_deadline_exceeded"),
+            (tonic::Status::permission_denied("remote-secret").into(), "rpc_permission_denied"),
+            (
+                Error::Io(std::io::Error::new(std::io::ErrorKind::TimedOut, "remote-secret")),
+                "io_timeout",
+            ),
+        ] {
+            let class = scanner_publication_lease_error_class(&error);
+            assert_eq!(class.reason, Some(expected));
+            assert_eq!(class.reported_reason, None);
+            assert!(!format!("{class:?}").contains("remote-secret"));
+        }
     }
 
     #[test]

crates/ecstore/src/services/notification_sys.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use crate::bucket::lifecycle::tier_last_day_stats::DailyAllTierStats;
+use crate::cluster::rpc::peer_rest_client::{ScannerLeaseConnectionFailure, scanner_publication_lease_error_class};
 use crate::cluster::rpc::{
     PeerRestClient, ScannerDirtyUsageAcknowledgement, ScannerPeerActivity, ScannerPeerDirtyUsageSnapshot,
     ScannerPublicationLease, TierConfigReloadOutcome,
@@ -158,16 +159,132 @@ struct PeerAdminCache {
     last_server_success: Option<SystemTime>,
 }
 
-fn scanner_publication_lease_acquisition_error(host: &str, observational_only: bool, error: Error) -> Error {
+enum ScannerPublicationLeaseFailure {
+    DuplicateTarget {
+        peer: String,
+    },
+    PeerUnavailable {
+        peer: String,
+    },
+    RequestFailed {
+        peer: String,
+        purpose: &'static str,
+        detail: String,
+        reason: Option<&'static str>,
+        reported_reason: Option<&'static str>,
+    },
+}
+
+impl std::fmt::Debug for ScannerPublicationLeaseFailure {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str("ScannerPublicationLeaseFailure")
+    }
+}
+
+impl std::fmt::Display for ScannerPublicationLeaseFailure {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::DuplicateTarget { peer } => write!(formatter, "duplicate scanner publication lease target: {peer}"),
+            Self::PeerUnavailable { peer } => write!(formatter, "scanner publication lease peer {peer} is unavailable"),
+            Self::RequestFailed {
+                peer, purpose, detail, ..
+            } => write!(
+                formatter,
+                "scanner publication lease acquisition failed: peer={peer}, purpose={purpose}: {detail}"
+            ),
+        }
+    }
+}
+
+impl std::error::Error for ScannerPublicationLeaseFailure {}
+
+/// Bounded details from the target that actually failed lease acquisition.
+#[derive(Clone, Copy, Debug)]
+pub struct ScannerPublicationLeaseFailureInfo<'a> {
+    pub peer: Option<&'a str>,
+    pub kind: &'static str,
+    pub reason: Option<&'static str>,
+    pub reported_reason: Option<&'static str>,
+}
+
+fn scanner_publication_lease_target_peer(target: &str) -> Option<&str> {
+    if target.len() > 267 {
+        return None;
+    }
+    let (peer, parse_target) = if let Some(peer) = target.strip_prefix("http://").or_else(|| target.strip_prefix("https://")) {
+        (peer, std::borrow::Cow::Borrowed(target))
+    } else {
+        (target, std::borrow::Cow::Owned(format!("http://{target}")))
+    };
+    if peer.is_empty()
+        || peer.len() > 259
+        || peer.ends_with(':')
+        || !peer
+            .bytes()
+            .all(|byte| byte.is_ascii_alphanumeric() || b".-_:[]".contains(&byte))
+    {
+        return None;
+    }
+    let parsed = url::Url::parse(&parse_target).ok()?;
+    if !matches!(parsed.scheme(), "http" | "https")
+        || parsed.host().is_none()
+        || !parsed.username().is_empty()
+        || parsed.password().is_some()
+        || parsed.path() != "/"
+        || parsed.query().is_some()
+        || parsed.fragment().is_some()
+    {
+        return None;
+    }
+    Some(peer)
+}
+
+/// Return structured lease failure context without parsing remote error text.
+pub fn scanner_publication_lease_failure_info(error: &Error) -> Option<ScannerPublicationLeaseFailureInfo<'_>> {
+    let Error::Io(error) = error else { return None };
+    let failure = error.get_ref()?.downcast_ref::<ScannerPublicationLeaseFailure>()?;
+    let (peer, kind, reason, reported_reason) = match failure {
+        ScannerPublicationLeaseFailure::DuplicateTarget { peer } => (peer, "duplicate_target", None, None),
+        ScannerPublicationLeaseFailure::PeerUnavailable { peer } => (peer, "peer_unavailable", None, None),
+        ScannerPublicationLeaseFailure::RequestFailed {
+            peer,
+            reason,
+            reported_reason,
+            ..
+        } => (peer, "peer_request_failed", *reason, *reported_reason),
+    };
+    let safe_peer = scanner_publication_lease_target_peer(peer);
+    Some(ScannerPublicationLeaseFailureInfo {
+        peer: safe_peer,
+        kind,
+        reason,
+        reported_reason,
+    })
+}
+
+fn scanner_publication_lease_acquisition_error(
+    host: &str,
+    observational_only: bool,
+    error: Error,
+    connection_failure: Option<ScannerLeaseConnectionFailure>,
+) -> Error {
     let purpose = if observational_only {
         "observational"
     } else {
         "authoritative"
     };
     // Retry classification depends on the original error suffix.
-    Error::other(format!(
-        "scanner publication lease acquisition failed: peer={host}, purpose={purpose}: {error}"
-    ))
+    let class = scanner_publication_lease_error_class(&error);
+    Error::other(ScannerPublicationLeaseFailure::RequestFailed {
+        peer: host.to_owned(),
+        purpose,
+        detail: error.to_string(),
+        reason: match connection_failure {
+            Some(ScannerLeaseConnectionFailure::ChannelAcquisitionTimedOut) => Some("channel_acquisition_timed_out"),
+            None => class.reason,
+        },
+        reported_reason: class.reported_reason,
+    })
 }
 
 #[derive(Default)]
@@ -3161,7 +3278,7 @@ impl NotificationSys {
         targets.sort_by(|left, right| left.0.cmp(&right.0));
         for pair in targets.windows(2) {
             if pair[0].0 == pair[1].0 {
-                return Err(Error::other(format!("duplicate scanner publication lease target: {}", pair[0].0)));
+                return Err(Error::other(ScannerPublicationLeaseFailure::DuplicateTarget { peer: pair[0].0.clone() }));
             }
         }
 
@@ -3175,16 +3292,21 @@ impl NotificationSys {
                 .cloned()
             else {
                 let _ = self.release_scanner_publication_leases(grants).await;
-                return Err(Error::other(format!("scanner publication lease peer {host} is unavailable")));
+                return Err(Error::other(ScannerPublicationLeaseFailure::PeerUnavailable { peer: host }));
             };
             match client
-                .acquire_scanner_publication_lease(&session_id, generation, observational_only)
+                .acquire_scanner_publication_lease_with_connection_failure(&session_id, generation, observational_only)
                 .await
             {
                 Ok(lease) => grants.push(ScannerPublicationLeaseGrant { host, lease }),
-                Err(err) => {
+                Err((err, connection_failure)) => {
                     let _ = self.release_scanner_publication_leases(grants).await;
-                    return Err(scanner_publication_lease_acquisition_error(&host, observational_only, err));
+                    return Err(scanner_publication_lease_acquisition_error(
+                        &host,
+                        observational_only,
+                        err,
+                        connection_failure,
+                    ));
                 }
             }
         }
@@ -4185,20 +4307,243 @@ mod tests {
 
     #[test]
     fn scanner_lease_acquisition_error_identifies_peer_and_preserves_retry_suffix() {
+        use crate::cluster::rpc::peer_rest_client::scanner_publication_lease_remote_rejection;
+
         for (observational_only, purpose) in [(false, "authoritative"), (true, "observational")] {
-            let reason = "scanner publication lease is blocked by pending namespace commit";
-            let error = scanner_publication_lease_acquisition_error(
-                "peer-a.example.invalid:9000",
-                observational_only,
-                Error::other(reason),
-            )
-            .to_string();
-            assert!(error.contains("peer-a.example.invalid:9000"), "missing peer: {error}");
-            assert!(error.contains(purpose), "missing purpose: {error}");
-            assert!(error.ends_with(reason), "retry classification must retain the original suffix");
+            let reason = "scanner publication lease capacity is exhausted";
+            let source = scanner_publication_lease_remote_rejection(format!("Io error: {reason}"));
+            let expected = format!(
+                "Io error: scanner publication lease acquisition failed: peer=peer-a.example.invalid:9000, purpose={purpose}: {source}"
+            );
+            let error =
+                scanner_publication_lease_acquisition_error("peer-a.example.invalid:9000", observational_only, source, None);
+            let info = scanner_publication_lease_failure_info(&error).expect("structured failure source");
+            assert_eq!(info.peer, Some("peer-a.example.invalid:9000"));
+            assert_eq!(info.kind, "peer_request_failed");
+            assert_eq!(info.reason, None);
+            assert_eq!(info.reported_reason, Some("capacity_exhausted"));
+            assert_eq!(error.to_string(), expected);
+            assert!(
+                error.to_string().ends_with(reason),
+                "retry classification must retain the original suffix"
+            );
+            let cloned = error.clone();
+            assert_eq!(cloned.to_string(), expected);
+            assert!(
+                scanner_publication_lease_failure_info(&cloned).is_none(),
+                "a cloned generic I/O error must not invent lost structured context"
+            );
         }
     }
 
+    #[test]
+    fn scanner_lease_failure_info_uses_the_actual_target_and_rejects_unbounded_peers() {
+        for (failure, expected_display, expected_kind) in [
+            (
+                ScannerPublicationLeaseFailure::DuplicateTarget {
+                    peer: "peer-a:9000".to_owned(),
+                },
+                "Io error: duplicate scanner publication lease target: peer-a:9000",
+                "duplicate_target",
+            ),
+            (
+                ScannerPublicationLeaseFailure::PeerUnavailable {
+                    peer: "peer-a:9000".to_owned(),
+                },
+                "Io error: scanner publication lease peer peer-a:9000 is unavailable",
+                "peer_unavailable",
+            ),
+        ] {
+            let error = Error::other(failure);
+            let info = scanner_publication_lease_failure_info(&error).expect("structured failure source");
+            assert_eq!(info.peer, Some("peer-a:9000"));
+            assert_eq!(info.kind, expected_kind);
+            assert_eq!(info.reason, None);
+            assert_eq!(info.reported_reason, None);
+            assert_eq!(error.to_string(), expected_display);
+        }
+
+        let forged =
+            Error::other("scanner publication lease acquisition failed: peer=forged:9000, purpose=authoritative: rejected");
+        assert!(
+            scanner_publication_lease_failure_info(&forged).is_none(),
+            "raw error text must not establish peer identity"
+        );
+        for invalid in [
+            "peer-a:9000\nforged",
+            "user:secret@peer-a:9000",
+            "http://user:secret@peer-a:9000",
+            "http://@peer-a:9000",
+            "http://peer-a:65536",
+            "https://peer-a:bad",
+            "https://peer-a:",
+            "http://[2001:db8::1",
+            "http://[2001:db8::1]:bad",
+            "https://peer-a:9000/path",
+            "https://peer-a:9000?token=secret",
+            "https://peer-a:9000#secret",
+            "https://peer-a%2eexample:9000",
+            "ftp://peer-a:9000",
+            "",
+        ] {
+            let error = scanner_publication_lease_acquisition_error(invalid, false, Error::other("remote-secret"), None);
+            let info = scanner_publication_lease_failure_info(&error).expect("structured failure source");
+            assert_eq!(info.peer, None, "invalid target must not reach logs");
+            assert_eq!(info.reported_reason, None, "unrecognized remote text must not become a reason");
+            assert!(!format!("{error:?}").contains("remote-secret"), "Debug must not expose remote error text");
+        }
+        let oversized = format!("https://{}:9000", "a".repeat(260));
+        let error = scanner_publication_lease_acquisition_error(&oversized, false, Error::other("remote-secret"), None);
+        assert_eq!(scanner_publication_lease_failure_info(&error).and_then(|info| info.peer), None);
+    }
+
+    #[test]
+    fn scanner_lease_failure_info_accepts_real_grid_host_urls() {
+        use crate::cluster::rpc::peer_rest_client::scanner_publication_lease_remote_rejection;
+        use crate::layout::endpoint::Endpoint;
+
+        for (endpoint, expected) in [
+            ("http://peer-a.example:9000/export", "peer-a.example:9000"),
+            ("https://peer-b.example:9443/export", "peer-b.example:9443"),
+            ("http://[2001:db8::1]:9000/export", "[2001:db8::1]:9000"),
+            ("https://peer-c.example/export", "peer-c.example"),
+        ] {
+            let target = Endpoint::try_from(endpoint).expect("valid endpoint").grid_host();
+            let source = scanner_publication_lease_remote_rejection(
+                "Io error: scanner publication lease capacity is exhausted".to_owned(),
+            );
+            let expected_display =
+                format!("Io error: scanner publication lease acquisition failed: peer={target}, purpose=observational: {source}");
+            let error = scanner_publication_lease_acquisition_error(&target, true, source, None);
+            let info = scanner_publication_lease_failure_info(&error).expect("structured failure source");
+            assert_eq!(info.peer, Some(expected));
+            assert_eq!(info.reason, None);
+            assert_eq!(info.reported_reason, Some("capacity_exhausted"));
+            assert_eq!(error.to_string(), expected_display);
+            assert!(error.to_string().ends_with("scanner publication lease capacity is exhausted"));
+        }
+    }
+
+    #[test]
+    fn scanner_lease_failure_info_preserves_typed_connection_failures_without_logging_details() {
+        for (source, reason) in [
+            (Error::RemoteClientUnavailable("remote-secret".to_owned()), "client_unavailable"),
+            (Error::Timeout, "timeout"),
+            (tonic::Status::unavailable("remote-secret").into(), "rpc_unavailable"),
+            (tonic::Status::deadline_exceeded("remote-secret").into(), "rpc_deadline_exceeded"),
+        ] {
+            let expected = format!(
+                "Io error: scanner publication lease acquisition failed: peer=https://peer-a:9000, purpose=authoritative: {source}"
+            );
+            let error = scanner_publication_lease_acquisition_error("https://peer-a:9000", false, source, None);
+            let info = scanner_publication_lease_failure_info(&error).expect("structured failure source");
+            assert_eq!(info.peer, Some("peer-a:9000"));
+            assert_eq!(info.kind, "peer_request_failed");
+            assert_eq!(info.reason, Some(reason));
+            assert_eq!(info.reported_reason, None);
+            assert_eq!(error.to_string(), expected);
+            assert!(!format!("{info:?}").contains("remote-secret"));
+        }
+
+        let spoofed = scanner_publication_lease_acquisition_error(
+            "https://peer-a:9000",
+            false,
+            Error::other("peer returned a different scanner publication lease generation"),
+            None,
+        );
+        let info = scanner_publication_lease_failure_info(&spoofed).expect("structured failure source");
+        assert_eq!(info.reason, None);
+        assert_eq!(info.reported_reason, None);
+    }
+
+    #[test]
+    fn scanner_lease_channel_deadline_preserves_public_error_and_diagnostic() {
+        const CHILD_MARKER: &str = "RUSTFS_TEST_SCANNER_LEASE_CHANNEL_DEADLINE_CHILD";
+        if std::env::var_os(CHILD_MARKER).is_none() {
+            // Isolate the zero deadline from concurrently running tests.
+            let module = module_path!()
+                .strip_prefix(concat!(env!("CARGO_CRATE_NAME"), "::"))
+                .unwrap_or(module_path!());
+            let name = format!("{module}::scanner_lease_channel_deadline_preserves_public_error_and_diagnostic");
+            let output = std::process::Command::new(std::env::current_exe().expect("test executable"))
+                .args(["--exact", &name, "--nocapture"])
+                .env(CHILD_MARKER, "1")
+                .env(rustfs_config::ENV_INTERNODE_CONNECT_TIMEOUT_SECS, "0")
+                .env("NO_PROXY", "127.0.0.1,localhost")
+                .env("no_proxy", "127.0.0.1,localhost")
+                .output()
+                .expect("deadline test subprocess");
+            assert!(
+                output.status.success() && String::from_utf8_lossy(&output.stdout).contains("1 passed"),
+                "deadline subprocess failed: {} {}",
+                String::from_utf8_lossy(&output.stdout),
+                String::from_utf8_lossy(&output.stderr)
+            );
+            return;
+        }
+
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        runtime.block_on(async {
+            // Keep the address unique to this isolated subprocess.
+            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("local listener");
+            let address = listener.local_addr().expect("listener address");
+            let target = format!("https://{address}");
+            let make_client = || {
+                PeerRestClient::new(
+                    XHost {
+                        name: "127.0.0.1".to_owned(),
+                        port: address.port(),
+                        is_port_set: true,
+                    },
+                    target.clone(),
+                )
+            };
+            // Block the real channel cache read so the acquisition deadline
+            // wins independently of OS-specific connection or TLS errors.
+            let cache_read_blocker = rustfs_common::block_connection_cache_reads_for_test().await;
+            let public_error = match make_client().get_client().await {
+                Ok(_) => panic!("a stalled connection must exceed the channel deadline"),
+                Err(error) => error,
+            };
+            assert!(matches!(&public_error, Error::RemoteClientUnavailable(_)));
+            assert!(
+                public_error
+                    .to_string()
+                    .contains("internode channel acquisition deadline elapsed")
+            );
+
+            let mut notification = NotificationSys::new(EndpointServerPools::default()).await;
+            notification.peer_clients.push(Some(make_client()));
+            let lease_error = match notification
+                .acquire_scanner_publication_leases(vec![(target.clone(), "session".into(), 3)])
+                .await
+            {
+                Ok(_) => panic!("a stalled connection cannot grant a lease"),
+                Err(error) => error,
+            };
+            assert_eq!(
+                lease_error.to_string(),
+                format!(
+                    "Io error: scanner publication lease acquisition failed: peer={target}, purpose=authoritative: {public_error}"
+                )
+            );
+            let info = scanner_publication_lease_failure_info(&lease_error).expect("structured lease failure");
+            let expected_peer = address.to_string();
+            assert_eq!(info.peer, Some(expected_peer.as_str()));
+            assert_eq!(info.reason, Some("channel_acquisition_timed_out"));
+            assert_eq!(info.reported_reason, None);
+            assert_eq!(
+                scanner_peer_transport_error_message_is_retryable(&lease_error.to_string()),
+                scanner_peer_transport_error_message_is_retryable(&public_error.to_string())
+            );
+            drop(cache_read_blocker);
+            drop(listener);
+        });
+    }
+
     fn ring(total_size: u64) -> LastDayTierStats {
         let mut stats = LastDayTierStats::default();
         stats.add_stats(TierStats {

crates/scanner/Cargo.toml

@@ -107,6 +107,10 @@ bytes.workspace = true
 hex-simd.workspace = true
 
 [dev-dependencies]
+opentelemetry.workspace = true
+opentelemetry-appender-tracing = { workspace = true, features = ["experimental_span_attributes", "experimental_metadata_attributes"] }
+opentelemetry_sdk.workspace = true
+rustfs-common = { workspace = true, features = ["test-util"] }
 rustfs-heal.workspace = true
 rustfs-test-utils.workspace = true
 tracing-subscriber = { workspace = true, features = ["json", "env-filter", "time"] }

crates/scanner/src/scanner.rs

@@ -2082,7 +2082,7 @@ where
         None
     } else if let Some(notification_system) = storeapi.scanner_notification_system() {
         let lease_result = publication_plan
-            .await_leases(ctx, &cycle_budget, cycle_info.current, |observational_lease| {
+            .await_leases(ctx, &cycle_budget, cycle_info.current, leader_epoch, |observational_lease| {
                 let targets = remote_publication_lease_targets.clone();
                 let notification_system = Arc::clone(&notification_system);
                 async move {
@@ -3892,17 +3892,17 @@ impl ScannerCyclePublicationPlan {
         self != Self::Authoritative
     }
 
-    async fn await_leases<T, E, F, Fut>(
+    async fn await_leases<T, F, Fut>(
         self,
         parent: &CancellationToken,
         budget: &ScannerCycleBudget,
         cycle: u64,
+        leader_epoch: u64,
         mut acquire: F,
-    ) -> ScannerPublicationProofWait<T, E>
+    ) -> ScannerPublicationProofWait<T, EcstoreError>
     where
-        E: std::fmt::Display,
         F: FnMut(bool) -> Fut,
-        Fut: Future<Output = Result<T, E>>,
+        Fut: Future<Output = Result<T, EcstoreError>>,
     {
         // Exhausting the walk budget must not cancel the movement proof needed
         // to save partial progress. No usage update means namespace stability
@@ -3912,6 +3912,17 @@ impl ScannerCyclePublicationPlan {
         } else {
             budget.token()
         };
+        let started_at = Instant::now();
+        let publication_kind = match self {
+            Self::ProgressOnly => "progress_only",
+            Self::Observational => "observational",
+            Self::Authoritative => "authoritative",
+        };
+        let lease_purpose = if self.uses_observational_lease() {
+            "observational"
+        } else {
+            "authoritative"
+        };
         debug!(
             target: "rustfs::scanner",
             event = EVENT_SCANNER_CYCLE_STATE,
@@ -3919,30 +3930,66 @@ impl ScannerCyclePublicationPlan {
             subsystem = LOG_SUBSYSTEM_RUNTIME,
             state = "publication_proof_started",
             cycle,
+            leader_epoch,
             stage = "lease_acquire",
-            publication_kind = match self {
-                Self::ProgressOnly => "progress_only",
-                Self::Observational => "observational",
-                Self::Authoritative => "authoritative",
-            },
+            publication_kind,
+            lease_purpose,
             "Scanner is acquiring peer publication proof"
         );
         let proof = await_scanner_publication_proof(
             &proof_ctx,
-            cycle,
-            "lease_acquire",
+            ScannerPublicationProofContext {
+                cycle,
+                stage: "lease_acquire",
+                leader_epoch: Some(leader_epoch),
+                lease_purpose: Some(lease_purpose),
+            },
             || acquire(self.uses_observational_lease()),
-            |err| scanner_publication_lease_error_is_retryable(&err.to_string()),
+            |err| {
+                // The structured reason comes from local acquisition, not a peer report.
+                crate::storage_api::scan::scanner_publication_lease_failure_info(err)
+                    .is_some_and(|info| info.reason == Some("channel_acquisition_timed_out"))
+                    || scanner_publication_lease_error_is_retryable(&err.to_string())
+            },
+            crate::storage_api::scan::scanner_publication_lease_failure_info,
         );
-        if self == Self::ProgressOnly {
+        let (proof, timed_out) = if self == Self::ProgressOnly {
             // Object/directory budgets can expire with no runtime deadline.
             // Bound the control-plane drain independently of the walk budget.
-            tokio::time::timeout(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, proof)
-                .await
-                .unwrap_or(ScannerPublicationProofWait::Cancelled)
+            match tokio::time::timeout(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, proof).await {
+                Ok(proof) => (proof, false),
+                Err(_) => (ScannerPublicationProofWait::Cancelled, true),
+            }
         } else {
-            proof.await
-        }
+            (proof.await, false)
+        };
+        let outcome = if timed_out {
+            "timed_out"
+        } else {
+            match &proof {
+                ScannerPublicationProofWait::Ready(_) => "acquired",
+                ScannerPublicationProofWait::Rejected(_) => "rejected",
+                ScannerPublicationProofWait::Cancelled => "cancelled",
+            }
+        };
+        debug!(
+            target: "rustfs::scanner",
+            event = EVENT_SCANNER_CYCLE_STATE,
+            component = LOG_COMPONENT_SCANNER,
+            subsystem = LOG_SUBSYSTEM_RUNTIME,
+            state = "publication_lease_wait_finished",
+            cycle,
+            leader_epoch,
+            stage = "lease_acquire",
+            publication_kind,
+            lease_purpose,
+            outcome,
+            parent_cancelled_at_return = parent.is_cancelled(),
+            proof_token_cancelled_at_return = proof_ctx.is_cancelled(),
+            elapsed = ?started_at.elapsed(),
+            "Scanner peer publication lease wait finished"
+        );
+        proof
     }
 }
 

crates/scanner/src/scanner/activity.rs

@@ -14,7 +14,7 @@
 /// Cycle wake/backoff policy and scanner activity observation (probing, generations, topology digest).
 use super::*;
 use crate::storage_api::ScannerStorage;
-use crate::storage_api::scan::SCANNER_ACTIVITY_V6_PROTOCOL_VERSION;
+use crate::storage_api::scan::{SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, ScannerPublicationLeaseFailureInfo};
 use std::collections::HashSet;
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -239,7 +239,7 @@ pub(crate) fn scanner_publication_proof_retry_delay(consecutive_failures: u32) -
 }
 
 pub(crate) fn scanner_publication_activity_error_is_retryable(error: &str) -> bool {
-    crate::storage_api::scanner_peer_transport_error_message_is_retryable(error)
+    crate::storage_api::scan::scanner_peer_transport_error_message_is_retryable(error)
 }
 
 pub(crate) fn scanner_publication_lease_error_is_retryable(error: &str) -> bool {
@@ -256,12 +256,19 @@ pub(crate) enum ScannerPublicationProofWait<T, E> {
     Cancelled,
 }
 
+pub(crate) struct ScannerPublicationProofContext {
+    pub cycle: u64,
+    pub stage: &'static str,
+    pub leader_epoch: Option<u64>,
+    pub lease_purpose: Option<&'static str>,
+}
+
 pub(crate) async fn await_scanner_publication_proof<T, E, F, Fut, Retryable>(
     ctx: &CancellationToken,
-    cycle: u64,
-    stage: &'static str,
+    context: ScannerPublicationProofContext,
     mut proof: F,
     retryable: Retryable,
+    failure_info: for<'a> fn(&'a E) -> Option<ScannerPublicationLeaseFailureInfo<'a>>,
 ) -> ScannerPublicationProofWait<T, E>
 where
     F: FnMut() -> Fut,
@@ -269,6 +276,12 @@ where
     E: std::fmt::Display,
     Retryable: Fn(&E) -> bool,
 {
+    let ScannerPublicationProofContext {
+        cycle,
+        stage,
+        leader_epoch,
+        lease_purpose,
+    } = context;
     let started_at = Instant::now();
     let mut consecutive_failures = 0u32;
 
@@ -288,6 +301,8 @@ where
                         state = "publication_proof_recovered",
                         cycle,
                         stage,
+                        leader_epoch,
+                        lease_purpose,
                         attempts = consecutive_failures.saturating_add(1),
                         pending_duration = ?started_at.elapsed(),
                         "Scanner publication proof recovered"
@@ -296,6 +311,11 @@ where
                 return ScannerPublicationProofWait::Ready(value);
             }
             Err(err) if !retryable(&err) => {
+                let failure = failure_info(&err);
+                let peer = failure.and_then(|info| info.peer);
+                let failure_kind = failure.map(|info| info.kind);
+                let reason = failure.and_then(|info| info.reason);
+                let reported_reason = failure.and_then(|info| info.reported_reason);
                 warn!(
                     target: "rustfs::scanner",
                     event = EVENT_SCANNER_CYCLE_STATE,
@@ -304,7 +324,12 @@ where
                     state = "publication_proof_rejected",
                     cycle,
                     stage,
-                    error = %err,
+                    leader_epoch,
+                    lease_purpose,
+                    peer,
+                    failure_kind,
+                    reason,
+                    reported_reason,
                     "Scanner publication proof failed with a non-retryable cluster state"
                 );
                 return ScannerPublicationProofWait::Rejected(err);
@@ -312,6 +337,11 @@ where
             Err(err) => {
                 consecutive_failures = consecutive_failures.saturating_add(1);
                 let retry_delay = scanner_publication_proof_retry_delay(consecutive_failures);
+                let failure = failure_info(&err);
+                let peer = failure.and_then(|info| info.peer);
+                let failure_kind = failure.map(|info| info.kind);
+                let reason = failure.and_then(|info| info.reason);
+                let reported_reason = failure.and_then(|info| info.reported_reason);
                 if consecutive_failures == 1 || consecutive_failures.is_multiple_of(20) {
                     warn!(
                         target: "rustfs::scanner",
@@ -321,10 +351,15 @@ where
                         state = "publication_proof_pending",
                         cycle,
                         stage,
+                        leader_epoch,
+                        lease_purpose,
+                        peer,
+                        failure_kind,
+                        reason,
+                        reported_reason,
                         attempt = consecutive_failures,
                         retry_delay = ?retry_delay,
                         pending_duration = ?started_at.elapsed(),
-                        error = %err,
                         "Scanner is waiting for publication proof"
                     );
                 } else {
@@ -336,9 +371,14 @@ where
                         state = "publication_proof_retry",
                         cycle,
                         stage,
+                        leader_epoch,
+                        lease_purpose,
+                        peer,
+                        failure_kind,
+                        reason,
+                        reported_reason,
                         attempt = consecutive_failures,
                         retry_delay = ?retry_delay,
-                        error = %err,
                         "Scanner publication activity proof retry scheduled"
                     );
                 }
@@ -363,9 +403,18 @@ where
     F: FnMut() -> Fut,
     Fut: Future<Output = Result<ScannerActivitySnapshot, String>>,
 {
-    match await_scanner_publication_proof(ctx, cycle, stage, probe, |err: &String| {
-        scanner_publication_activity_error_is_retryable(err)
-    })
+    match await_scanner_publication_proof(
+        ctx,
+        ScannerPublicationProofContext {
+            cycle,
+            stage,
+            leader_epoch: None,
+            lease_purpose: None,
+        },
+        probe,
+        |err: &String| scanner_publication_activity_error_is_retryable(err),
+        |_| None,
+    )
     .await
     {
         ScannerPublicationProofWait::Ready(snapshot) => Ok(snapshot),

crates/scanner/src/scanner/tests.rs

@@ -9204,8 +9204,12 @@ async fn publication_lease_retries_pending_namespace_commit_until_it_clears() {
 
     let result = await_scanner_publication_proof(
         &ctx,
-        17,
-        "lease_acquire",
+        ScannerPublicationProofContext {
+            cycle: 17,
+            stage: "lease_acquire",
+            leader_epoch: None,
+            lease_purpose: None,
+        },
         move || {
             let attempt = proof_attempts.fetch_add(1, Ordering::SeqCst);
             async move {
@@ -9217,6 +9221,7 @@ async fn publication_lease_retries_pending_namespace_commit_until_it_clears() {
             }
         },
         |error| scanner_publication_lease_error_is_retryable(error),
+        |_| None,
     )
     .await;
 

crates/scanner/src/scanner/tests/cycle_budget_publication.rs

@@ -13,8 +13,629 @@
 // limitations under the License.
 
 use super::*;
+use crate::storage_api::scan::ScannerPublicationLeaseFailureInfo;
 use rustfs_utils::path::path_join_buf;
 
+#[derive(Clone)]
+struct LeaseLogWriter(Arc<std::sync::Mutex<Vec<u8>>>);
+
+impl std::io::Write for LeaseLogWriter {
+    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+        self.0.lock().expect("lease log mutex").extend_from_slice(bytes);
+        Ok(bytes.len())
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_lease_observations_keep_cycle_epoch_purpose_and_outcome() {
+    use tracing::instrument::WithSubscriber as _;
+
+    let output = Arc::new(std::sync::Mutex::new(Vec::new()));
+    let writer = LeaseLogWriter(Arc::clone(&output));
+    let subscriber = tracing_subscriber::fmt()
+        .json()
+        .with_max_level(tracing::Level::DEBUG)
+        .with_writer(move || writer.clone())
+        .with_ansi(false)
+        .without_time()
+        .finish();
+    let _interest_guard = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
+
+    async {
+        let ctx = CancellationToken::new();
+        let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
+        let ready = ScannerCyclePublicationPlan::ProgressOnly
+            .await_leases(&ctx, &budget, 501, 17, |_| async { Ok::<_, EcstoreError>(()) })
+            .await;
+        assert!(matches!(ready, ScannerPublicationProofWait::Ready(())));
+        let next_leader = ScannerCyclePublicationPlan::Observational
+            .await_leases(&ctx, &budget, 501, 22, |_| async { Ok::<_, EcstoreError>(()) })
+            .await;
+        assert!(matches!(next_leader, ScannerPublicationProofWait::Ready(())));
+        let rejected = ScannerCyclePublicationPlan::Observational
+            .await_leases(&ctx, &budget, 502, 18, |_| async {
+                Err::<(), _>(EcstoreError::other("protocol mismatch"))
+            })
+            .await;
+        assert!(matches!(rejected, ScannerPublicationProofWait::Rejected(_)));
+        ctx.cancel();
+        let cancelled = ScannerCyclePublicationPlan::Authoritative
+            .await_leases(&ctx, &budget, 503, 19, |_| async { Ok::<_, EcstoreError>(()) })
+            .await;
+        assert!(matches!(cancelled, ScannerPublicationProofWait::Cancelled));
+        let timeout_ctx = CancellationToken::new();
+        let timeout_budget = ScannerCycleBudget::new(&timeout_ctx, ScannerCycleBudgetConfig::default());
+        let timed_out = ScannerCyclePublicationPlan::ProgressOnly
+            .await_leases(&timeout_ctx, &timeout_budget, 504, 20, |_| async {
+                std::future::pending::<Result<(), EcstoreError>>().await
+            })
+            .await;
+        assert!(matches!(timed_out, ScannerPublicationProofWait::Cancelled));
+        let budget_ctx = CancellationToken::new();
+        let stopped_budget = ScannerCycleBudget::new(&budget_ctx, ScannerCycleBudgetConfig::default());
+        stopped_budget.cancel_for_runtime();
+        let budget_cancelled = ScannerCyclePublicationPlan::Observational
+            .await_leases(&budget_ctx, &stopped_budget, 505, 21, |_| async { Ok::<_, EcstoreError>(()) })
+            .await;
+        assert!(matches!(budget_cancelled, ScannerPublicationProofWait::Cancelled));
+    }
+    .with_subscriber(subscriber)
+    .await;
+
+    let bytes = output.lock().expect("lease log mutex").clone();
+    let logs = String::from_utf8(bytes).expect("lease logs are UTF-8");
+    let entries: Vec<serde_json::Value> = logs
+        .lines()
+        .map(|line| serde_json::from_str(line).expect("lease logs are JSON"))
+        .collect();
+    for (cycle, epoch, kind, purpose, outcome, parent_cancelled, proof_cancelled) in [
+        (501, 17, "progress_only", "observational", "acquired", false, false),
+        (502, 18, "observational", "observational", "rejected", false, false),
+        (503, 19, "authoritative", "authoritative", "cancelled", true, true),
+        (504, 20, "progress_only", "observational", "timed_out", false, false),
+        (505, 21, "observational", "observational", "cancelled", false, true),
+        (501, 22, "observational", "observational", "acquired", false, false),
+    ] {
+        let started: Vec<_> = entries
+            .iter()
+            .filter_map(|entry| entry.get("fields"))
+            .filter(|fields| {
+                fields["state"] == "publication_proof_started" && fields["cycle"] == cycle && fields["leader_epoch"] == epoch
+            })
+            .collect();
+        assert_eq!(started.len(), 1, "missing or duplicated started event for cycle {cycle}: {logs}");
+        assert_eq!(started[0]["leader_epoch"], epoch);
+        assert_eq!(started[0]["publication_kind"], kind);
+        assert_eq!(started[0]["lease_purpose"], purpose);
+        let finished: Vec<_> = entries
+            .iter()
+            .filter_map(|entry| entry.get("fields"))
+            .filter(|fields| {
+                fields["state"] == "publication_lease_wait_finished"
+                    && fields["cycle"] == cycle
+                    && fields["leader_epoch"] == epoch
+            })
+            .collect();
+        assert_eq!(finished.len(), 1, "missing or duplicated finished event for cycle {cycle}: {logs}");
+        let fields = finished[0];
+        assert_eq!(fields["leader_epoch"], epoch);
+        assert_eq!(fields["publication_kind"], kind);
+        assert_eq!(fields["lease_purpose"], purpose);
+        assert_eq!(fields["outcome"], outcome);
+        assert_eq!(fields["parent_cancelled_at_return"], parent_cancelled);
+        assert_eq!(fields["proof_token_cancelled_at_return"], proof_cancelled);
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_lease_retry_logs_identify_interleaved_epochs() {
+    use tracing::instrument::WithSubscriber as _;
+
+    for level in [tracing::Level::WARN, tracing::Level::INFO, tracing::Level::DEBUG] {
+        let output = Arc::new(std::sync::Mutex::new(Vec::new()));
+        let writer = LeaseLogWriter(Arc::clone(&output));
+        let subscriber = tracing_subscriber::fmt()
+            .json()
+            .flatten_event(true)
+            .with_env_filter(format!("off,rustfs::scanner={level}"))
+            .with_current_span(true)
+            .with_writer(move || writer.clone())
+            .with_ansi(false)
+            .without_time()
+            .finish();
+        let _interest_guard = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
+
+        exercise_interleaved_lease_waits().with_subscriber(subscriber).await;
+
+        let bytes = output.lock().expect("lease log mutex").clone();
+        let logs = String::from_utf8(bytes).expect("lease logs are UTF-8");
+        assert!(!logs.contains("remote-secret"), "untrusted error text must not enter logs");
+        assert!(!logs.contains("forged:9000"), "unstructured error text must not establish a peer");
+        let entries: Vec<serde_json::Value> = logs
+            .lines()
+            .map(|line| serde_json::from_str(line).expect("lease logs are JSON"))
+            .collect();
+        let find_lease_event = |state: &str| {
+            entries
+                .iter()
+                .position(|entry| entry["state"] == state && entry["stage"] == "lease_acquire")
+                .unwrap_or_else(|| panic!("missing {state} at {level}: {logs}"))
+        };
+        let pending = find_lease_event("publication_proof_pending");
+        let rejected = find_lease_event("publication_proof_rejected");
+        assert!(pending < rejected, "waits must interleave: {logs}");
+        for (index, epoch, purpose) in [(pending, 41, "observational"), (rejected, 42, "authoritative")] {
+            assert_eq!(entries[index]["cycle"], 601);
+            assert_eq!(entries[index]["leader_epoch"], epoch, "{logs}");
+            assert_eq!(entries[index]["lease_purpose"], purpose, "{logs}");
+        }
+        assert_eq!(entries[rejected]["peer"], "peer-a:9000", "{logs}");
+        assert_eq!(entries[rejected]["failure_kind"], "peer_unavailable", "{logs}");
+        let ipv6 = entries
+            .iter()
+            .find(|entry| entry["state"] == "publication_proof_rejected" && entry["cycle"] == 604)
+            .expect("IPv6 target failure should be logged");
+        assert_eq!(ipv6["peer"], "[2001:db8::1]:9443", "{logs}");
+        let secret_target = entries
+            .iter()
+            .find(|entry| entry["state"] == "publication_proof_rejected" && entry["cycle"] == 605)
+            .expect("credential-bearing target failure should be logged");
+        assert!(secret_target["peer"].is_null(), "credential-bearing target must not be logged");
+        assert!(!logs.contains("target-secret"), "target credentials must not enter logs");
+        let local = entries
+            .iter()
+            .find(|entry| entry["state"] == "publication_proof_rejected" && entry["cycle"] == 606)
+            .expect("local diagnostic should be logged");
+        assert_eq!(local["reason"], "invalid_proof");
+        assert!(local["reported_reason"].is_null());
+        let reported = entries
+            .iter()
+            .find(|entry| entry["state"] == "publication_proof_rejected" && entry["cycle"] == 607)
+            .expect("remote report should be logged");
+        assert!(reported["reason"].is_null());
+        assert_eq!(reported["reported_reason"], "generation_stale");
+        let timeout = entries
+            .iter()
+            .find(|entry| entry["state"] == "publication_proof_rejected" && entry["cycle"] == 608)
+            .expect("channel deadline should be logged");
+        assert_eq!(timeout["reason"], "channel_acquisition_timed_out");
+        assert!(timeout["reported_reason"].is_null());
+        assert!(!logs.contains("diagnostic-secret"), "error text must not enter logs");
+        if level != tracing::Level::WARN {
+            let recovered = find_lease_event("publication_proof_recovered");
+            assert!(rejected < recovered, "waits must interleave: {logs}");
+            assert_eq!(entries[recovered]["leader_epoch"], 41, "{logs}");
+            assert_eq!(entries[recovered]["lease_purpose"], "observational", "{logs}");
+        }
+        if level == tracing::Level::DEBUG {
+            let retry = find_lease_event("publication_proof_retry");
+            let recovered = find_lease_event("publication_proof_recovered");
+            assert!(rejected < retry && retry < recovered, "retry must remain interleaved: {logs}");
+            assert_eq!(entries[retry]["leader_epoch"], 41, "{logs}");
+            assert_eq!(entries[retry]["lease_purpose"], "observational", "{logs}");
+        }
+        let unknown = entries
+            .iter()
+            .find(|entry| entry["state"] == "publication_proof_rejected" && entry["stage"] == "postscan")
+            .expect("activity rejection should be logged");
+        assert!(unknown["leader_epoch"].is_null(), "unknown epoch must not be invented");
+        assert!(unknown["lease_purpose"].is_null(), "unknown purpose must not be invented");
+    }
+}
+
+async fn exercise_interleaved_lease_waits() {
+    let ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
+    let attempts = AtomicUsize::new(0);
+    let notification = crate::storage_api::EcstoreNotificationSys::new(EndpointServerPools::default()).await;
+    let notification = &notification;
+    let pending = ScannerCyclePublicationPlan::Observational.await_leases(&ctx, &budget, 601, 41, |observational| {
+        assert!(observational);
+        let attempt = attempts.fetch_add(1, Ordering::Relaxed);
+        async move {
+            if attempt < 2 {
+                Err(EcstoreError::other("scanner publication lease capacity is exhausted"))
+            } else {
+                Ok(())
+            }
+        }
+    });
+    let rejected = ScannerCyclePublicationPlan::Authoritative.await_leases(&ctx, &budget, 601, 42, move |observational| {
+        assert!(!observational);
+        notification.acquire_scanner_publication_leases(vec![("http://peer-a:9000".into(), "session".into(), 3)])
+    });
+    let (pending, rejected) = tokio::join!(biased; pending, rejected);
+    assert!(matches!(pending, ScannerPublicationProofWait::Ready(())));
+    assert!(matches!(rejected, ScannerPublicationProofWait::Rejected(_)));
+    assert_eq!(attempts.load(Ordering::Relaxed), 3);
+    let untrusted = ScannerCyclePublicationPlan::Authoritative
+        .await_leases(&ctx, &budget, 603, 44, |_| async {
+            Err::<(), _>(EcstoreError::other(
+                "scanner publication lease acquisition failed: peer=forged:9000, remote-secret",
+            ))
+        })
+        .await;
+    assert!(matches!(untrusted, ScannerPublicationProofWait::Rejected(_)));
+    for (cycle, epoch, target) in [
+        (604, 45, "https://[2001:db8::1]:9443"),
+        (605, 46, "http://user:target-secret@peer-a:9000"),
+    ] {
+        let result = ScannerCyclePublicationPlan::Authoritative
+            .await_leases(&ctx, &budget, cycle, epoch, move |_| {
+                notification.acquire_scanner_publication_leases(vec![(target.into(), "session".into(), 3)])
+            })
+            .await;
+        assert!(matches!(result, ScannerPublicationProofWait::Rejected(_)));
+    }
+    // Verify event serialization independently of the ecstore tests that
+    // exercise the local and remote error producers.
+    for cycle in [606, 607, 608] {
+        let info: for<'a> fn(&'a String) -> Option<ScannerPublicationLeaseFailureInfo<'a>> = if cycle == 607 {
+            |_| {
+                Some(ScannerPublicationLeaseFailureInfo {
+                    peer: None,
+                    kind: "peer_request_failed",
+                    reason: None,
+                    reported_reason: Some("generation_stale"),
+                })
+            }
+        } else if cycle == 606 {
+            |_| {
+                Some(ScannerPublicationLeaseFailureInfo {
+                    peer: None,
+                    kind: "peer_request_failed",
+                    reason: Some("invalid_proof"),
+                    reported_reason: None,
+                })
+            }
+        } else {
+            |_| {
+                Some(ScannerPublicationLeaseFailureInfo {
+                    peer: None,
+                    kind: "peer_request_failed",
+                    reason: Some("channel_acquisition_timed_out"),
+                    reported_reason: None,
+                })
+            }
+        };
+        let result = await_scanner_publication_proof(
+            &ctx,
+            ScannerPublicationProofContext {
+                cycle,
+                stage: "lease_acquire",
+                leader_epoch: Some(47),
+                lease_purpose: Some("authoritative"),
+            },
+            || async { Err::<(), _>("diagnostic-secret".to_owned()) },
+            |_| false,
+            info,
+        )
+        .await;
+        assert!(matches!(result, ScannerPublicationProofWait::Rejected(_)));
+    }
+    let unknown =
+        await_scanner_publication_activity(&ctx, 601, "postscan", || async { Err("protocol mismatch".to_string()) }).await;
+    assert!(unknown.is_err());
+}
+
+#[derive(Clone, Debug, Default)]
+struct CapturedOtelLogs(Arc<std::sync::Mutex<Vec<opentelemetry_sdk::logs::SdkLogRecord>>>);
+
+impl opentelemetry_sdk::logs::LogProcessor for CapturedOtelLogs {
+    fn emit(&self, data: &mut opentelemetry_sdk::logs::SdkLogRecord, _: &opentelemetry::InstrumentationScope) {
+        self.0.lock().expect("OTLP log mutex").push(data.clone());
+    }
+
+    fn force_flush(&self) -> opentelemetry_sdk::error::OTelSdkResult {
+        Ok(())
+    }
+
+    fn shutdown_with_timeout(&self, _: std::time::Duration) -> opentelemetry_sdk::error::OTelSdkResult {
+        Ok(())
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_lease_events_include_context_in_otlp_without_span_export() {
+    use opentelemetry::logs::AnyValue;
+    use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
+    use tracing::{Instrument as _, instrument::WithSubscriber as _};
+    use tracing_subscriber::prelude::*;
+
+    for level in [tracing::Level::WARN, tracing::Level::INFO, tracing::Level::DEBUG] {
+        let captured = CapturedOtelLogs::default();
+        let provider = opentelemetry_sdk::logs::SdkLoggerProvider::builder()
+            .with_log_processor(captured.clone())
+            .build();
+        let subscriber = tracing_subscriber::registry()
+            .with(tracing_subscriber::EnvFilter::new(format!("off,rustfs::scanner={level}")))
+            .with(OpenTelemetryTracingBridge::new(&provider));
+        let _interest_guard = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
+        async {
+            // The bridge is configured like production: no tracing exporter or
+            // span attribute copying. A parent span must not supply these fields.
+            let unrelated = tracing::warn_span!(
+                target: "rustfs::scanner",
+                "unrelated_parent",
+                leader_epoch = 999,
+                lease_purpose = "unrelated"
+            );
+            exercise_interleaved_lease_waits().instrument(unrelated).await;
+        }
+        .with_subscriber(subscriber)
+        .await;
+
+        let records = captured.0.lock().expect("OTLP log mutex").clone();
+        let attribute = |record: &opentelemetry_sdk::logs::SdkLogRecord, name: &str| {
+            record
+                .attributes_iter()
+                .find(|(key, _)| key.as_str() == name)
+                .map(|(_, value)| value.clone())
+        };
+        let find = |state: &'static str, cycle: i64, stage: &'static str| {
+            records
+                .iter()
+                .find(|record| {
+                    attribute(record, "state") == Some(AnyValue::from(state))
+                        && attribute(record, "cycle") == Some(AnyValue::Int(cycle))
+                        && attribute(record, "stage") == Some(AnyValue::from(stage))
+                })
+                .unwrap_or_else(|| panic!("missing {state} at {level}"))
+        };
+        let pending = find("publication_proof_pending", 601, "lease_acquire");
+        let rejected = find("publication_proof_rejected", 601, "lease_acquire");
+        for (record, epoch, purpose) in [(pending, 41, "observational"), (rejected, 42, "authoritative")] {
+            assert_eq!(attribute(record, "leader_epoch"), Some(AnyValue::Int(epoch)));
+            assert_eq!(attribute(record, "lease_purpose"), Some(AnyValue::from(purpose)));
+        }
+        assert_eq!(attribute(rejected, "peer"), Some(AnyValue::from("peer-a:9000")));
+        assert_eq!(attribute(rejected, "failure_kind"), Some(AnyValue::from("peer_unavailable")));
+        let ipv6 = find("publication_proof_rejected", 604, "lease_acquire");
+        assert_eq!(attribute(ipv6, "peer"), Some(AnyValue::from("[2001:db8::1]:9443")));
+        let secret_target = find("publication_proof_rejected", 605, "lease_acquire");
+        assert!(attribute(secret_target, "peer").is_none());
+        let local = find("publication_proof_rejected", 606, "lease_acquire");
+        assert_eq!(attribute(local, "reason"), Some(AnyValue::from("invalid_proof")));
+        assert!(attribute(local, "reported_reason").is_none());
+        let reported = find("publication_proof_rejected", 607, "lease_acquire");
+        assert!(attribute(reported, "reason").is_none());
+        assert_eq!(attribute(reported, "reported_reason"), Some(AnyValue::from("generation_stale")));
+        let timeout = find("publication_proof_rejected", 608, "lease_acquire");
+        assert_eq!(attribute(timeout, "reason"), Some(AnyValue::from("channel_acquisition_timed_out")));
+        assert!(attribute(timeout, "reported_reason").is_none());
+        if level != tracing::Level::WARN {
+            let recovered = find("publication_proof_recovered", 601, "lease_acquire");
+            assert_eq!(attribute(recovered, "leader_epoch"), Some(AnyValue::Int(41)));
+            assert_eq!(attribute(recovered, "lease_purpose"), Some(AnyValue::from("observational")));
+        }
+        if level == tracing::Level::DEBUG {
+            let retry = find("publication_proof_retry", 601, "lease_acquire");
+            assert_eq!(attribute(retry, "leader_epoch"), Some(AnyValue::Int(41)));
+            assert_eq!(attribute(retry, "lease_purpose"), Some(AnyValue::from("observational")));
+        }
+        let unknown = find("publication_proof_rejected", 601, "postscan");
+        assert!(attribute(unknown, "leader_epoch").is_none());
+        assert!(attribute(unknown, "lease_purpose").is_none());
+        let untrusted = find("publication_proof_rejected", 603, "lease_acquire");
+        assert!(attribute(untrusted, "peer").is_none());
+        assert!(!format!("{records:?}").contains("remote-secret"));
+        assert!(!format!("{records:?}").contains("forged:9000"));
+        assert!(!format!("{records:?}").contains("target-secret"));
+        assert!(!format!("{records:?}").contains("diagnostic-secret"));
+    }
+}
+
+#[tokio::test]
+async fn dropped_lease_wait_does_not_report_a_finished_outcome() {
+    use tracing::instrument::WithSubscriber as _;
+
+    let output = Arc::new(std::sync::Mutex::new(Vec::new()));
+    let writer = LeaseLogWriter(Arc::clone(&output));
+    let subscriber = tracing_subscriber::fmt()
+        .json()
+        .with_max_level(tracing::Level::DEBUG)
+        .with_writer(move || writer.clone())
+        .with_ansi(false)
+        .without_time()
+        .finish();
+    let _interest_guard = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
+
+    async {
+        let ctx = CancellationToken::new();
+        let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
+        let entered = AtomicBool::new(false);
+        {
+            let proof = ScannerCyclePublicationPlan::Observational.await_leases(&ctx, &budget, 602, 43, |_| {
+                entered.store(true, Ordering::Relaxed);
+                std::future::pending::<Result<(), EcstoreError>>()
+            });
+            tokio::pin!(proof);
+            assert!(futures::poll!(&mut proof).is_pending());
+            assert!(entered.load(Ordering::Relaxed));
+        }
+        assert!(!ctx.is_cancelled());
+        assert!(!budget.token().is_cancelled());
+    }
+    .with_subscriber(subscriber)
+    .await;
+
+    let bytes = output.lock().expect("lease log mutex").clone();
+    let logs = String::from_utf8(bytes).expect("lease logs are UTF-8");
+    let entries: Vec<serde_json::Value> = logs
+        .lines()
+        .map(|line| serde_json::from_str(line).expect("lease logs are JSON"))
+        .collect();
+    let started = entries
+        .iter()
+        .filter(|entry| entry["fields"]["state"] == "publication_proof_started" && entry["fields"]["cycle"] == 602)
+        .count();
+    let finished = entries
+        .iter()
+        .filter(|entry| entry["fields"]["state"] == "publication_lease_wait_finished" && entry["fields"]["cycle"] == 602)
+        .count();
+    assert_eq!(started, 1, "the wait was polled: {logs}");
+    assert_eq!(finished, 0, "a dropped future has no observed outcome: {logs}");
+}
+
+#[test]
+fn publication_lease_retries_local_channel_deadline_and_stops_on_cancel() {
+    const CHILD_MARKER: &str = "RUSTFS_TEST_SCANNER_LEASE_DIAL_RETRY_CHILD";
+    if std::env::var_os(CHILD_MARKER).is_none() {
+        let module = module_path!()
+            .strip_prefix(concat!(env!("CARGO_CRATE_NAME"), "::"))
+            .unwrap_or(module_path!());
+        let name = format!("{module}::publication_lease_retries_local_channel_deadline_and_stops_on_cancel");
+        let output = std::process::Command::new(std::env::current_exe().expect("test executable"))
+            .args(["--exact", &name, "--nocapture"])
+            .env(CHILD_MARKER, "1")
+            .env(rustfs_config::ENV_INTERNODE_CONNECT_TIMEOUT_SECS, "0")
+            .env("NO_PROXY", "127.0.0.1,localhost")
+            .env("no_proxy", "127.0.0.1,localhost")
+            .output()
+            .expect("channel deadline test subprocess");
+        assert!(
+            output.status.success() && String::from_utf8_lossy(&output.stdout).contains("1 passed"),
+            "channel deadline subprocess failed: {} {}",
+            String::from_utf8_lossy(&output.stdout),
+            String::from_utf8_lossy(&output.stderr)
+        );
+        return;
+    }
+
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .start_paused(true)
+        .build()
+        .expect("test runtime");
+    runtime.block_on(async {
+        use crate::storage_api::scan::EcstorePeerRestClient;
+        use rustfs_utils::net::XHost;
+
+        async fn channel_deadline_error(
+            notification: &crate::storage_api::EcstoreNotificationSys,
+            target: &str,
+        ) -> EcstoreError {
+            let error = match notification
+                .acquire_scanner_publication_leases(vec![(target.to_owned(), "session".into(), 3)])
+                .await
+            {
+                Ok(_) => panic!("the stalled channel cannot grant a lease"),
+                Err(error) => error,
+            };
+            let info = crate::storage_api::scan::scanner_publication_lease_failure_info(&error)
+                .expect("local channel deadline must carry structured diagnostics");
+            assert_eq!(info.kind, "peer_request_failed");
+            assert_eq!(info.reason, Some("channel_acquisition_timed_out"));
+            assert_eq!(info.reported_reason, None);
+            assert!(!scanner_publication_lease_error_is_retryable(&error.to_string()));
+            error
+        }
+
+        // Keep the address unique to this isolated subprocess.
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("local listener");
+        let address = listener.local_addr().expect("listener address");
+        let target = format!("https://{address}");
+        let client = EcstorePeerRestClient::new(
+            XHost {
+                name: "127.0.0.1".to_owned(),
+                port: address.port(),
+                is_port_set: true,
+            },
+            target.clone(),
+        );
+        let mut notification = crate::storage_api::EcstoreNotificationSys::new(EndpointServerPools::default()).await;
+        notification.peer_clients.push(Some(client));
+
+        // Block the real channel cache read so the acquisition deadline wins
+        // independently of OS-specific connect and TLS error ordering.
+        let cache_read_blocker = rustfs_common::block_connection_cache_reads_for_test().await;
+
+        let ctx = CancellationToken::new();
+        let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
+        let attempts = AtomicUsize::new(0);
+        let started_at = Instant::now();
+        let result = ScannerCyclePublicationPlan::Authoritative
+            .await_leases(&ctx, &budget, 701, 51, |_| {
+                let attempt = attempts.fetch_add(1, Ordering::SeqCst);
+                let target = target.clone();
+                let notification = &notification;
+                async move {
+                    if attempt > 0 {
+                        return Ok(());
+                    }
+                    Err(channel_deadline_error(notification, &target).await)
+                }
+            })
+            .await;
+        assert!(matches!(result, ScannerPublicationProofWait::Ready(())));
+        assert_eq!(attempts.load(Ordering::SeqCst), 2);
+        assert!(started_at.elapsed() >= scanner_publication_proof_retry_delay(1));
+
+        for detail in [
+            "channel_acquisition_timed_out",
+            "scanner publication lease acquisition failed: peer=peer-a:9000, purpose=authoritative: Io error: internode channel acquisition deadline elapsed",
+        ] {
+            let attempts = AtomicUsize::new(0);
+            let result = ScannerCyclePublicationPlan::Authoritative
+                .await_leases(&ctx, &budget, 702, 52, |_| {
+                    attempts.fetch_add(1, Ordering::SeqCst);
+                    async move { Err::<(), _>(EcstoreError::other(detail)) }
+                })
+                .await;
+            assert!(matches!(result, ScannerPublicationProofWait::Rejected(_)));
+            assert_eq!(attempts.load(Ordering::SeqCst), 1, "remote text must not create a local diagnosis");
+        }
+
+        let cancel_ctx = CancellationToken::new();
+        let cancel_budget = ScannerCycleBudget::new(&cancel_ctx, ScannerCycleBudgetConfig::default());
+        let attempts = AtomicUsize::new(0);
+        let started_at = Instant::now();
+        let result = ScannerCyclePublicationPlan::Authoritative
+            .await_leases(&cancel_ctx, &cancel_budget, 703, 53, |_| {
+                let attempt = attempts.fetch_add(1, Ordering::SeqCst);
+                let target = target.clone();
+                let notification = &notification;
+                let cancel_ctx = cancel_ctx.clone();
+                async move {
+                    assert_eq!(attempt, 0, "cancellation must stop the scheduled retry");
+                    let error = channel_deadline_error(notification, &target).await;
+                    tokio::spawn(async move {
+                        tokio::time::sleep(Duration::from_secs(1)).await;
+                        cancel_ctx.cancel();
+                    });
+                    Err::<(), _>(error)
+                }
+            })
+            .await;
+        assert!(matches!(result, ScannerPublicationProofWait::Cancelled));
+        assert_eq!(attempts.load(Ordering::SeqCst), 1);
+        assert!(started_at.elapsed() < scanner_publication_proof_retry_delay(1));
+
+        let progress_ctx = CancellationToken::new();
+        let progress_budget = ScannerCycleBudget::new(&progress_ctx, ScannerCycleBudgetConfig::default());
+        let attempts = AtomicUsize::new(0);
+        let started_at = Instant::now();
+        let result = ScannerCyclePublicationPlan::ProgressOnly
+            .await_leases(&progress_ctx, &progress_budget, 704, 54, |observational| {
+                assert!(observational);
+                attempts.fetch_add(1, Ordering::SeqCst);
+                let target = target.clone();
+                let notification = &notification;
+                async move { Err::<(), _>(channel_deadline_error(notification, &target).await) }
+            })
+            .await;
+        assert!(matches!(result, ScannerPublicationProofWait::Cancelled));
+        assert_eq!(attempts.load(Ordering::SeqCst), 1);
+        assert!(started_at.elapsed() < scanner_publication_proof_retry_delay(1));
+        drop(cache_read_blocker);
+        drop(listener);
+    });
+}
+
 #[tokio::test]
 #[serial]
 async fn exhausted_budget_still_proves_peer_movement_without_namespace_authority() {
@@ -33,7 +654,7 @@ async fn exhausted_budget_still_proves_peer_movement_without_namespace_authority
     let calls = AtomicUsize::new(0);
 
     let proof = plan
-        .await_leases(&ctx, &budget, 352, |observational| {
+        .await_leases(&ctx, &budget, 352, 19, |observational| {
             calls.fetch_add(1, Ordering::Relaxed);
             store.acquire_scanner_publication_lease_with_purpose(generation, Duration::from_secs(60), observational)
         })
@@ -57,7 +678,7 @@ async fn exhausted_budget_still_proves_peer_movement_without_namespace_authority
     assert!(store.release_scanner_publication_lease(token).await);
 
     let stale = plan
-        .await_leases(&ctx, &budget, 352, |observational| {
+        .await_leases(&ctx, &budget, 352, 19, |observational| {
             store.acquire_scanner_publication_lease_with_purpose(generation + 1, Duration::from_secs(60), observational)
         })
         .await;
@@ -71,7 +692,7 @@ async fn exhausted_budget_still_proves_peer_movement_without_namespace_authority
     assert!(store.scanner_data_movement_pause_status().await.paused);
     let current_generation = store.scanner_data_movement_generation();
     let moving = plan
-        .await_leases(&ctx, &budget, 352, |observational| {
+        .await_leases(&ctx, &budget, 352, 19, |observational| {
             store.acquire_scanner_publication_lease_with_purpose(current_generation, Duration::from_secs(60), observational)
         })
         .await;
@@ -101,15 +722,21 @@ async fn emitted_updates_keep_budget_cancellation_and_publication_purpose() {
         let ctx = CancellationToken::new();
         let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
         let ready = plan
-            .await_leases(&ctx, &budget, 352, |observational| async move { Ok::<_, String>(observational) })
+            .await_leases(
+                &ctx,
+                &budget,
+                352,
+                19,
+                |observational| async move { Ok::<_, EcstoreError>(observational) },
+            )
             .await;
         assert!(matches!(ready, ScannerPublicationProofWait::Ready(observational) if observational == plan.is_observational()));
         budget.cancel_for_runtime();
         let calls = AtomicUsize::new(0);
         let cancelled = plan
-            .await_leases(&ctx, &budget, 352, |_| {
+            .await_leases(&ctx, &budget, 352, 19, |_| {
                 calls.fetch_add(1, Ordering::Relaxed);
-                async { Ok::<(), String>(()) }
+                async { Ok::<(), EcstoreError>(()) }
             })
             .await;
         assert!(matches!(cancelled, ScannerPublicationProofWait::Cancelled));
@@ -124,9 +751,9 @@ async fn progress_proof_honors_parent_cancellation() {
     ctx.cancel();
     let calls = AtomicUsize::new(0);
     let proof = ScannerCyclePublicationPlan::ProgressOnly
-        .await_leases(&ctx, &budget, 352, |_| {
+        .await_leases(&ctx, &budget, 352, 19, |_| {
             calls.fetch_add(1, Ordering::Relaxed);
-            async { Ok::<_, String>(()) }
+            async { Ok::<_, EcstoreError>(()) }
         })
         .await;
     assert!(matches!(proof, ScannerPublicationProofWait::Cancelled));
@@ -149,13 +776,13 @@ async fn progress_proof_is_bounded_without_a_runtime_budget() {
         let calls = AtomicUsize::new(0);
         let started = Instant::now();
         let proof = ScannerCyclePublicationPlan::ProgressOnly
-            .await_leases(&ctx, &budget, 352, |_| {
+            .await_leases(&ctx, &budget, 352, 19, |_| {
                 calls.fetch_add(1, Ordering::Relaxed);
                 async move {
                     if stalled {
                         std::future::pending::<()>().await;
                     }
-                    Err::<(), _>("scanner publication lease capacity is exhausted")
+                    Err::<(), _>(EcstoreError::other("scanner publication lease capacity is exhausted"))
                 }
             })
             .await;

crates/scanner/src/storage_api.rs

@@ -109,7 +109,6 @@ pub(crate) use rustfs_ecstore::api::layout::{
     SetupType as EcstoreSetupType,
 };
 pub(crate) use rustfs_ecstore::api::notification::NotificationSys as EcstoreNotificationSys;
-pub(crate) use rustfs_ecstore::api::notification::scanner_peer_transport_error_message_is_retryable;
 pub(crate) use rustfs_ecstore::api::object::{
     SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, ScannerPublicationCommitScope, ScannerPublicationCommitState,
 };
@@ -365,6 +364,12 @@ pub(crate) mod scan {
         SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE,
         SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES,
     };
+    pub(crate) use rustfs_ecstore::api::notification::{
+        ScannerPublicationLeaseFailureInfo, scanner_peer_transport_error_message_is_retryable,
+        scanner_publication_lease_failure_info,
+    };
+    #[cfg(test)]
+    pub(crate) use rustfs_ecstore::api::rpc::PeerRestClient as EcstorePeerRestClient;
 }
 
 pub(crate) mod scanner_io {