Run scoped multipart lock diagnostics on Windows

.github/workflows/windows-filesystem.yml

@@ -1,172 +1,26 @@
-# 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: Multipart Lock Windows 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: multipart-lock-windows-${{ 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
+  scoped-tests:
     runs-on: windows-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
+      - name: Setup Rust
         uses: ./.github/actions/setup
         with:
           rust-version: stable
@@ -174,51 +28,21 @@ jobs:
           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: Run only the named diagnostic unit tests
+        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
-        env:
-          RUSTFS_ACCESS_KEY: rustfsadmin
-          RUSTFS_SECRET_KEY: rustfsadmin
-        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-obs telemetry::local::tests::multipart_timeout_event_does_not_inherit_verbose_span_fields
+          run_one rustfs-obs telemetry::local::tests::multipart_timeout_target_keeps_the_existing_error_module_filter
+          run_one rustfs-obs telemetry::local::tests::multipart_timeout_root_event_does_not_inherit_otlp_span_fields
+          run_one rustfs error::tests::multipart_create_timeout_classifies_only_the_exact_known_resources
+          run_one rustfs error::tests::multipart_create_timeout_keeps_nested_and_untrusted_sources_distinct
+          run_one rustfs error::tests::multipart_create_timeout_event_is_visible_at_warn_and_does_not_expose_source
+          run_one rustfs storage::access::tests::multipart_timeout_observes_only_the_authorized_final_error
+          run_one rustfs app::multipart_usecase::tests::create_multipart_observes_the_actual_object_lock_timeout

crates/obs/src/lib.rs

@@ -81,6 +81,10 @@ pub use metrics::{
 };
 pub use telemetry::{HTTP_SERVER_LOG_TARGET, OtelGuard, Recorder};
 
+/// Target reserved for the bounded multipart lock-timeout event.
+/// The local JSON formatter must not attach request span fields to this event.
+pub const MULTIPART_CREATE_TIMEOUT_LOG_TARGET: &str = "rustfs::error::multipart_create_timeout";
+
 // Dial9 Tokio runtime telemetry
 // Re-export dial9 types at crate root level for easier access
 pub use telemetry::dial9;

crates/obs/src/telemetry/local.rs

@@ -181,11 +181,16 @@ fn resolve_log_cleanup_interval_seconds(config: &OtelConfig) -> u64 {
 #[derive(Clone, Debug)]
 struct RequestIdJsonFormat<T> {
     inner: Format<Json, T>,
+    multipart_timeout: Format<Json, T>,
 }
 
-impl<T> RequestIdJsonFormat<T> {
+impl<T: Clone> RequestIdJsonFormat<T> {
     fn new(inner: Format<Json, T>) -> Self {
-        Self { inner }
+        let multipart_timeout = inner.clone().with_current_span(false).with_span_list(false);
+        Self {
+            inner,
+            multipart_timeout,
+        }
     }
 }
 
@@ -201,6 +206,10 @@ where
         event: &tracing::Event<'_>,
     ) -> fmt::Result {
         let mut buffer = String::new();
+        if event.metadata().target() == crate::MULTIPART_CREATE_TIMEOUT_LOG_TARGET {
+            self.multipart_timeout.format_event(ctx, Writer::new(&mut buffer), event)?;
+            return writer.write_str(&buffer);
+        }
         self.inner.format_event(ctx, Writer::new(&mut buffer), event)?;
 
         let request_id = span_scope_request_id(ctx, event);
@@ -858,6 +867,156 @@ mod tests {
         .expect("log output should be valid UTF-8")
     }
 
+    #[test]
+    fn multipart_timeout_event_does_not_inherit_verbose_span_fields() {
+        for (level, span_events) in [
+            ("debug", FmtSpan::NONE),
+            ("debug", FmtSpan::CLOSE),
+            ("debug", FmtSpan::FULL),
+            ("trace", FmtSpan::NONE),
+            ("trace", FmtSpan::CLOSE),
+            ("trace", FmtSpan::FULL),
+        ] {
+            let writer = SharedWriter::default();
+            let layer = build_json_log_layer(writer.clone(), false, span_events, true);
+            let subscriber = Registry::default()
+                .with(tracing_subscriber::EnvFilter::new(level))
+                .with(layer);
+
+            tracing::subscriber::with_default(subscriber, || {
+                let sensitive = tracing::error_span!(
+                    "request",
+                    uri = "/bucket/LEAK-key?X-Amz-Signature=LEAK-signature",
+                    credential = "LEAK-credential"
+                );
+                let _guard = sensitive.enter();
+                tracing::error!(
+                    target: crate::MULTIPART_CREATE_TIMEOUT_LOG_TARGET,
+                    parent: None,
+                    event = "multipart_create_lock_timeout",
+                    request_id = "b8172a15-a428-4b42-9cef-1f4a7b96874e",
+                    resource_class = "request_object",
+                    "lock timeout"
+                );
+                tracing::error!(event = "unrelated_event", "normal logging still has its span");
+            });
+
+            let bytes = writer.inner.lock().expect("shared writer lock").clone();
+            let output = String::from_utf8(bytes).expect("UTF-8 logs");
+            let rows = output
+                .lines()
+                .map(|line| serde_json::from_str::<Value>(line).expect("JSON log"))
+                .collect::<Vec<_>>();
+            let event = rows
+                .iter()
+                .find(|row| row["event"] == "multipart_create_lock_timeout")
+                .expect("timeout event");
+            assert_eq!(event["request_id"], "b8172a15-a428-4b42-9cef-1f4a7b96874e");
+            assert!(event.get("span").is_none());
+            assert!(event.get("spans").is_none());
+            assert!(!event.to_string().contains("LEAK-"));
+            let ordinary = rows
+                .iter()
+                .find(|row| row["event"] == "unrelated_event")
+                .expect("ordinary event");
+            assert!(
+                ordinary.to_string().contains("LEAK-key"),
+                "other events keep their configured span context"
+            );
+        }
+    }
+
+    #[test]
+    fn multipart_timeout_target_keeps_the_existing_error_module_filter() {
+        let writer = SharedWriter::default();
+        let layer = build_json_log_layer(writer.clone(), false, FmtSpan::NONE, true);
+        let subscriber = Registry::default()
+            .with(tracing_subscriber::EnvFilter::new("off,rustfs::error=warn"))
+            .with(layer);
+        tracing::subscriber::with_default(subscriber, || {
+            tracing::error!(
+                target: crate::MULTIPART_CREATE_TIMEOUT_LOG_TARGET,
+                parent: None,
+                event = "multipart_create_lock_timeout",
+                "lock timeout"
+            );
+        });
+        let bytes = writer.inner.lock().expect("shared writer lock").clone();
+        let output = String::from_utf8(bytes).expect("UTF-8 logs");
+        let event: Value = serde_json::from_str(output.trim()).expect("filtered timeout event");
+        assert_eq!(event["event"], "multipart_create_lock_timeout");
+    }
+
+    #[test]
+    fn multipart_timeout_root_event_does_not_inherit_otlp_span_fields() {
+        use opentelemetry::InstrumentationScope;
+        use opentelemetry_appender_tracing::layer::{OpenTelemetryTracingBridge, TracingSpanAttributes};
+        use opentelemetry_sdk::{
+            error::OTelSdkResult,
+            logs::{LogProcessor, SdkLogRecord, SdkLoggerProvider},
+        };
+
+        #[derive(Clone, Debug, Default)]
+        struct Capture(Arc<Mutex<Vec<SdkLogRecord>>>);
+
+        impl LogProcessor for Capture {
+            fn emit(&self, record: &mut SdkLogRecord, _scope: &InstrumentationScope) {
+                self.0.lock().expect("captured records").push(record.clone());
+            }
+
+            fn force_flush(&self) -> OTelSdkResult {
+                Ok(())
+            }
+
+            fn shutdown_with_timeout(&self, _timeout: std::time::Duration) -> OTelSdkResult {
+                Ok(())
+            }
+        }
+
+        for copy_span_fields in [false, true] {
+            let capture = Capture::default();
+            let provider = SdkLoggerProvider::builder().with_log_processor(capture.clone()).build();
+            let bridge = if copy_span_fields {
+                OpenTelemetryTracingBridge::builder(&provider)
+                    .with_tracing_span_attributes(TracingSpanAttributes::all())
+                    .build()
+            } else {
+                OpenTelemetryTracingBridge::new(&provider)
+            };
+            let subscriber = Registry::default().with(bridge);
+            tracing::subscriber::with_default(subscriber, || {
+                let sensitive = tracing::error_span!(
+                    "request",
+                    uri = "/bucket/LEAK-key?X-Amz-Signature=LEAK-signature",
+                    credential = "LEAK-credential"
+                );
+                let _guard = sensitive.enter();
+                tracing::error!(
+                    target: crate::MULTIPART_CREATE_TIMEOUT_LOG_TARGET,
+                    parent: None,
+                    event = "multipart_create_lock_timeout",
+                    request_id = "b8172a15-a428-4b42-9cef-1f4a7b96874e",
+                    resource_class = "request_object",
+                    "lock timeout"
+                );
+                tracing::error!(event = "unrelated_event", "ordinary event");
+            });
+            let records = capture.0.lock().expect("captured records");
+            let timeout = records
+                .iter()
+                .find(|record| record.target().map(|target| target.as_ref()) == Some(crate::MULTIPART_CREATE_TIMEOUT_LOG_TARGET))
+                .expect("timeout event");
+            let output = format!("{timeout:?}");
+            assert!(output.contains("b8172a15-a428-4b42-9cef-1f4a7b96874e"));
+            assert!(!output.contains("LEAK-"));
+            let ordinary = records
+                .iter()
+                .find(|record| record.target().map(|target| target.as_ref()) != Some(crate::MULTIPART_CREATE_TIMEOUT_LOG_TARGET))
+                .expect("ordinary event");
+            assert_eq!(format!("{ordinary:?}").contains("LEAK-key"), copy_span_fields);
+        }
+    }
+
     fn render_json_log_with_request_span() -> Value {
         let writer = SharedWriter::default();
         let layer = build_json_log_layer(writer.clone(), false, FmtSpan::NONE, false);

rustfs/src/app/multipart_usecase.rs

@@ -17,7 +17,8 @@
 use super::storage_api::multipart_usecase::ECStore;
 use super::storage_api::multipart_usecase::access::{
     TableDataPlaneListAccess, apply_bucket_generation_guard, apply_copy_source_bucket_generation_guard,
-    has_bypass_governance_header, replication_request_authorized, retained_table_data_plane_publication_guards,
+    has_bypass_governance_header, replication_request_authorized, request_context_from_req,
+    retained_table_data_plane_publication_guards,
 };
 use super::storage_api::multipart_usecase::bucket::quota::checker::QuotaChecker;
 use super::storage_api::multipart_usecase::bucket::{
@@ -92,7 +93,7 @@ use crate::auth::{
     reject_presigned_put_max_content_length_for_other_operation,
 };
 use crate::capacity::record_capacity_write;
-use crate::error::ApiError;
+use crate::error::{ApiError, MultipartCreateBoundary, log_multipart_create_timeout};
 use crate::table_catalog;
 #[cfg(test)]
 use bytes::Bytes;
@@ -972,6 +973,24 @@ impl DefaultMultipartUsecase {
     pub async fn execute_create_multipart_upload(
         &self,
         req: S3Request<CreateMultipartUploadInput>,
+    ) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
+        let result = self.execute_create_multipart_upload_inner(&req).await;
+        if let Err(error) = &result {
+            let context = request_context_from_req(&req);
+            log_multipart_create_timeout(
+                error,
+                MultipartCreateBoundary::Execute,
+                context.as_ref().map(|context| context.request_id.as_str()),
+                &req.input.bucket,
+                &req.input.key,
+            );
+        }
+        result
+    }
+
+    async fn execute_create_multipart_upload_inner(
+        &self,
+        req: &S3Request<CreateMultipartUploadInput>,
     ) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
         let multipart_max_total_object_size = parse_presigned_multipart_max_total_object_size(
             &req.headers,
@@ -983,10 +1002,9 @@ impl DefaultMultipartUsecase {
             req.uri.query(),
             req.extensions.get::<VerifiedPresignedRequest>().is_some(),
         )?;
-        let helper =
-            OperationHelper::new(&req, EventName::ObjectCreatedCreateMultipartUpload, S3Operation::CreateMultipartUpload)
-                .suppress_event();
-        let replication_authorized = replication_request_authorized(&req);
+        let helper = OperationHelper::new(req, EventName::ObjectCreatedCreateMultipartUpload, S3Operation::CreateMultipartUpload)
+            .suppress_event();
+        let replication_authorized = replication_request_authorized(req);
         let CreateMultipartUploadInput {
             bucket,
             key,
@@ -1064,7 +1082,7 @@ impl DefaultMultipartUsecase {
 
         // The session data key is generated here, so this is where a multipart upload is held
         // to the KMS key it names. Parts and the completion reuse the resulting envelope.
-        let session_principal = SseKmsPrincipal::from_request(&req);
+        let session_principal = SseKmsPrincipal::from_request(req);
         let encryption_request = PrepareEncryptionRequest {
             bucket: &bucket,
             key: &key,
@@ -1139,7 +1157,7 @@ impl DefaultMultipartUsecase {
             put_opts_with_replication_authorization(&bucket, &key, version_id, &req.headers, metadata, replication_authorized)
                 .await
                 .map_err(ApiError::from)?;
-        apply_bucket_generation_guard(&req, &bucket, &mut opts)?;
+        apply_bucket_generation_guard(req, &bucket, &mut opts)?;
 
         let dsc =
             must_replicate_object(&bucket, &key, &mt2, "".to_string(), opts.delete_marker_replication_status(), opts.clone())
@@ -2684,6 +2702,71 @@ mod tests {
         assert_eq!(s3_op_total(S3Operation::AbortMultipartUpload), before + 1);
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn create_multipart_observes_the_actual_object_lock_timeout() {
+        use crate::app::storage_api::multipart_usecase::request_context::RequestContext;
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+        use crate::app::storage_api::test::contract::namespace::NamespaceLocking as _;
+        use crate::error::multipart_timeout_test_log::Capture;
+        use tracing::instrument::WithSubscriber as _;
+
+        let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+        let ambient = crate::app::gating_test_env::shared_gating_ambient().await;
+        let context = Arc::new(AppContext::new(Arc::clone(&store), ambient.iam(), ambient.kms()));
+        let bucket = format!("multipart-timeout-object-{}", Uuid::new_v4());
+        let key = "LEAK-private-key";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+        let _guard = store
+            .new_ns_lock(&bucket, key)
+            .await
+            .expect("create namespace lock")
+            .get_write_lock(Duration::from_secs(5))
+            .await
+            .expect("hold object lock");
+        let input = CreateMultipartUploadInput::builder()
+            .bucket(bucket.clone())
+            .key(key.to_string())
+            .build()
+            .expect("multipart input");
+        let mut req = build_request(input, Method::POST);
+        let request_context = RequestContext::from_external_headers(&req.headers);
+        let request_id = request_context.request_id.clone();
+        req.extensions.insert(request_context);
+
+        let capture = Capture::default();
+        let subscriber = tracing_subscriber::fmt()
+            .json()
+            .flatten_event(true)
+            .with_current_span(true)
+            .with_span_list(false)
+            .with_max_level(tracing::Level::WARN)
+            .with_writer(capture.clone())
+            .finish();
+        let error = DefaultMultipartUsecase::with_context(Some(context))
+            .execute_create_multipart_upload(req)
+            .with_subscriber(subscriber)
+            .await
+            .expect_err("the held object lock must time out");
+        assert_eq!(error.code(), &S3ErrorCode::ServiceUnavailable);
+        assert_eq!(error.headers().unwrap()["x-rustfs-error-cause"], "namespace_lock_timeout");
+        let rows = capture
+            .rows()
+            .into_iter()
+            .filter(|row| row["event"] == "multipart_create_lock_timeout")
+            .collect::<Vec<_>>();
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0]["boundary"], "execute");
+        assert_eq!(rows[0]["request_id"], request_id);
+        assert_eq!(rows[0]["resource_class"], "request_object");
+        assert_eq!(rows[0]["timeout_source"], "direct");
+        assert!(!rows[0].to_string().contains(&bucket));
+        assert!(!rows[0].to_string().contains(key));
+    }
+
     #[tokio::test]
     async fn execute_create_multipart_upload_rejects_invalid_storage_class() {
         let input = CreateMultipartUploadInput::builder()

rustfs/src/app/storage_api.rs

@@ -298,8 +298,8 @@ pub(crate) mod access {
         authorize_request, bucket_config_mutation_incarnation, delete_object_authorize_action, has_bypass_governance_header,
         load_bucket_generation_from_store, log_list_buckets_iam_implicit_deny, odm_read_generation,
         prepare_list_buckets_iam_authorization, prepare_odm_read_generation, recursive_force_delete_has_authenticated_caller,
-        replication_request_authorized, req_info_mut, req_info_ref, retained_table_data_plane_publication_guards,
-        set_requested_object_lock_retain_until,
+        replication_request_authorized, req_info_mut, req_info_ref, request_context_from_req,
+        retained_table_data_plane_publication_guards, set_requested_object_lock_retain_until,
     };
 }
 

rustfs/src/error.rs

@@ -656,6 +656,99 @@ fn service_unavailable_cause(source: &(dyn std::error::Error + 'static)) -> Opti
     timeout.or(cause)
 }
 
+#[derive(Clone, Copy)]
+pub(crate) enum MultipartCreateBoundary {
+    Access,
+    Execute,
+}
+
+struct MultipartCreateTimeout {
+    resource_class: &'static str,
+    source_kind: &'static str,
+    timeout_ms: u64,
+}
+
+fn multipart_timeout_resource_class(resource: &str, bucket: &str, key: &str) -> &'static str {
+    if bucket.is_empty() {
+        return "unknown";
+    }
+
+    let request_resource = resource
+        .strip_suffix("@latest")
+        .and_then(|path| path.strip_prefix(bucket))
+        .and_then(|path| path.strip_prefix('/'));
+    let lifecycle = request_resource == Some("");
+    let transaction = resource
+        .strip_prefix(".rustfs.sys/bucket-targets/")
+        .and_then(|path| path.strip_suffix("/transaction.lock@latest"))
+        == Some(bucket);
+    let object = if key.is_empty() {
+        false
+    } else {
+        let encoded = rustfs_utils::path::encode_dir_object_ref(key);
+        let identity = rustfs_utils::path::object_name_disk_identity(encoded.as_ref());
+        request_resource == Some(identity.as_ref())
+    };
+
+    // A formatted key may be ambiguous; do not choose a class in that case.
+    match (lifecycle, transaction, object) {
+        (true, false, false) => "bucket_lifecycle",
+        (false, true, false) => "bucket_transaction",
+        (false, false, true) => "request_object",
+        _ => "unknown",
+    }
+}
+
+fn multipart_create_timeout_diagnostic(error: &S3Error, bucket: &str, key: &str) -> Option<MultipartCreateTimeout> {
+    if error.code() != &S3ErrorCode::ServiceUnavailable || error.status_code() != Some(StatusCode::SERVICE_UNAVAILABLE) {
+        return None;
+    }
+    let source = error.source()?;
+    let direct = matches!(source.downcast_ref::<StorageError>(), Some(StorageError::Lock(LockError::Timeout { .. })))
+        || matches!(source.downcast_ref::<LockError>(), Some(LockError::Timeout { .. }));
+    error_chain_find(source, |source| match source.downcast_ref::<LockError>() {
+        Some(LockError::Timeout { resource, timeout }) => Some(MultipartCreateTimeout {
+            resource_class: multipart_timeout_resource_class(resource, bucket, key),
+            source_kind: if direct { "direct" } else { "nested" },
+            timeout_ms: u64::try_from(timeout.as_millis()).unwrap_or(u64::MAX),
+        }),
+        _ => None,
+    })
+}
+
+pub(crate) fn log_multipart_create_timeout(
+    error: &S3Error,
+    boundary: MultipartCreateBoundary,
+    request_id: Option<&str>,
+    bucket: &str,
+    key: &str,
+) {
+    let Some(diagnostic) = multipart_create_timeout_diagnostic(error, bucket, key) else {
+        return;
+    };
+    let request_id = request_id
+        .filter(|value| uuid::Uuid::parse_str(value).is_ok())
+        .unwrap_or("unknown");
+    let boundary = match boundary {
+        MultipartCreateBoundary::Access => "access",
+        MultipartCreateBoundary::Execute => "execute",
+    };
+    tracing::error!(
+        target: rustfs_obs::MULTIPART_CREATE_TIMEOUT_LOG_TARGET,
+        parent: None,
+        event = "multipart_create_lock_timeout",
+        component = "storage",
+        subsystem = "object_lock",
+        state = "failed",
+        boundary,
+        request_id,
+        resource_class = diagnostic.resource_class,
+        timeout_source = diagnostic.source_kind,
+        timeout_budget_ms = diagnostic.timeout_ms,
+        "Multipart creation returned a lock timeout source"
+    );
+}
+
 impl From<ApiError> for S3Error {
     fn from(err: ApiError) -> Self {
         let status = custom_error_status(&err.code);
@@ -899,6 +992,51 @@ impl From<QuotaError> for ApiError {
     }
 }
 
+#[cfg(test)]
+pub(crate) mod multipart_timeout_test_log {
+    use std::{
+        io,
+        io::Write,
+        sync::{Arc, Mutex},
+    };
+    use tracing_subscriber::fmt::MakeWriter;
+
+    #[derive(Clone, Default)]
+    pub(crate) struct Capture(Arc<Mutex<Vec<u8>>>);
+
+    pub(crate) struct Writer(Arc<Mutex<Vec<u8>>>);
+
+    impl Write for Writer {
+        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+            self.0.lock().expect("log capture").extend_from_slice(buf);
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'a> MakeWriter<'a> for Capture {
+        type Writer = Writer;
+
+        fn make_writer(&'a self) -> Self::Writer {
+            Writer(self.0.clone())
+        }
+    }
+
+    impl Capture {
+        pub(crate) fn rows(&self) -> Vec<serde_json::Value> {
+            let data = self.0.lock().expect("log capture").clone();
+            String::from_utf8(data)
+                .expect("UTF-8 logs")
+                .lines()
+                .map(|line| serde_json::from_str(line).expect("JSON log"))
+                .collect()
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -910,6 +1048,133 @@ mod tests {
     use s3s::{S3Error, S3ErrorCode};
     use std::io::{Error as IoError, ErrorKind};
 
+    fn multipart_timeout_error(resource: &str) -> S3Error {
+        ApiError::from(StorageError::Lock(LockError::timeout(resource, std::time::Duration::from_secs(5)))).into()
+    }
+
+    #[test]
+    fn multipart_create_timeout_classifies_only_the_exact_known_resources() {
+        for (resource, key, expected) in [
+            ("bucket/@latest", "private-key", "bucket_lifecycle"),
+            ("bucket/private-key@latest", "private-key", "request_object"),
+            ("bucket/dir__XLDIR__@latest", "dir/", "request_object"),
+            (
+                ".rustfs.sys/bucket-targets/bucket/transaction.lock@latest",
+                "private-key",
+                "bucket_transaction",
+            ),
+            (".rustfs.sys/bucket-targets/other/transaction.lock@latest", "private-key", "unknown"),
+            ("bucket/private-key@older", "private-key", "unknown"),
+            ("bucket/other@latest", "private-key", "unknown"),
+        ] {
+            let error = multipart_timeout_error(resource);
+            let diagnostic = multipart_create_timeout_diagnostic(&error, "bucket", key).expect("typed timeout");
+            assert_eq!(diagnostic.resource_class, expected, "{resource}");
+            assert_eq!(diagnostic.source_kind, "direct");
+            assert_eq!(diagnostic.timeout_ms, 5_000);
+        }
+
+        let key = "a\\b";
+        let encoded = rustfs_utils::path::object_name_disk_identity(key);
+        let error = multipart_timeout_error(&format!("bucket/{encoded}@latest"));
+        assert_eq!(
+            multipart_create_timeout_diagnostic(&error, "bucket", key)
+                .unwrap()
+                .resource_class,
+            "request_object"
+        );
+
+        let error = multipart_timeout_error(".rustfs.sys/bucket-targets/bucket/transaction.lock@latest");
+        assert_eq!(
+            multipart_create_timeout_diagnostic(&error, ".rustfs.sys", "bucket-targets/bucket/transaction.lock")
+                .unwrap()
+                .resource_class,
+            "request_object"
+        );
+        let ambiguous = multipart_timeout_error(".rustfs.sys/bucket-targets/.rustfs.sys/transaction.lock@latest");
+        assert_eq!(
+            multipart_create_timeout_diagnostic(&ambiguous, ".rustfs.sys", "bucket-targets/.rustfs.sys/transaction.lock")
+                .unwrap()
+                .resource_class,
+            "unknown"
+        );
+    }
+
+    #[test]
+    fn multipart_create_timeout_keeps_nested_and_untrusted_sources_distinct() {
+        let nested = S3Error::from(ApiError {
+            code: S3ErrorCode::ServiceUnavailable,
+            message: "unavailable".into(),
+            source: Some(Box::new(IoError::other(LockError::Network {
+                message: "LEAK-network-detail".into(),
+                source: Box::new(LockError::timeout("bucket/private-key@latest", std::time::Duration::from_secs(5))),
+            }))),
+        });
+        let diagnostic = multipart_create_timeout_diagnostic(&nested, "bucket", "private-key").expect("nested timeout");
+        assert_eq!(diagnostic.resource_class, "request_object");
+        assert_eq!(diagnostic.source_kind, "nested");
+
+        let forged = S3Error::from(ApiError {
+            code: S3ErrorCode::ServiceUnavailable,
+            message: "unavailable".into(),
+            source: Some(Box::new(IoError::other(
+                "Lock acquisition timeout for resource 'bucket/private-key@latest'",
+            ))),
+        });
+        assert!(multipart_create_timeout_diagnostic(&forged, "bucket", "private-key").is_none());
+        let mut denied = multipart_timeout_error("bucket/private-key@latest");
+        denied.set_code(S3ErrorCode::AccessDenied);
+        assert!(multipart_create_timeout_diagnostic(&denied, "bucket", "private-key").is_none());
+    }
+
+    #[test]
+    fn multipart_create_timeout_event_is_visible_at_warn_and_does_not_expose_source() {
+        use crate::error::multipart_timeout_test_log::Capture;
+
+        let capture = Capture::default();
+        let subscriber = tracing_subscriber::fmt()
+            .json()
+            .flatten_event(true)
+            .with_current_span(false)
+            .with_span_list(false)
+            .with_max_level(tracing::Level::WARN)
+            .with_writer(capture.clone())
+            .finish();
+        let error = multipart_timeout_error("private-bucket/LEAK-private-key@latest");
+        let request_id = "b8172a15-a428-4b42-9cef-1f4a7b96874e";
+        tracing::subscriber::with_default(subscriber, || {
+            log_multipart_create_timeout(
+                &error,
+                MultipartCreateBoundary::Execute,
+                Some(request_id),
+                "private-bucket",
+                "LEAK-private-key",
+            );
+            log_multipart_create_timeout(
+                &error,
+                MultipartCreateBoundary::Access,
+                Some("LEAK-invalid-request-id"),
+                "private-bucket",
+                "LEAK-private-key",
+            );
+        });
+        let rows = capture.rows();
+        assert_eq!(rows.len(), 2);
+        assert_eq!(rows[0]["event"], "multipart_create_lock_timeout");
+        assert_eq!(rows[0]["boundary"], "execute");
+        assert_eq!(rows[0]["request_id"], request_id);
+        assert_eq!(rows[0]["resource_class"], "request_object");
+        assert_eq!(rows[0]["timeout_budget_ms"], 5_000);
+        let output = rows[0].to_string();
+        assert!(!output.contains("private-bucket"));
+        assert!(!output.contains("LEAK-private-key"));
+        assert!(!output.contains("LEAK-parent-span"));
+        assert_eq!(rows[1]["request_id"], "unknown");
+        assert!(!rows[1].to_string().contains("LEAK-invalid-request-id"));
+        assert_eq!(error.headers().unwrap()["x-rustfs-error-cause"], "namespace_lock_timeout");
+        assert_eq!(error.headers().unwrap()[http::header::RETRY_AFTER], "1");
+    }
+
     fn error_chain_has_type<T>(err: &(dyn std::error::Error + 'static)) -> bool
     where
         T: std::error::Error + 'static,

rustfs/src/storage/access.rs

@@ -22,7 +22,7 @@ use crate::auth::{
     get_request_auth_type_with_query, get_session_token, parse_presigned_multipart_max_total_object_size,
     parse_presigned_put_max_content_length, reject_unsigned_amz_headers_on_sigv4_request,
 };
-use crate::error::{ApiError, normalize_read_quorum_error};
+use crate::error::{ApiError, MultipartCreateBoundary, log_multipart_create_timeout, normalize_read_quorum_error};
 use crate::license::license_check;
 use crate::shared_types::RemoteAddr;
 use crate::storage::request_context::RequestContext;
@@ -462,6 +462,7 @@ static OBJECT_GENERATION_TEST_HOOK: OnceLock<std::sync::Mutex<Option<ObjectGener
 #[cfg(test)]
 tokio::task_local! {
     static PUT_GENERATION_READ_QUORUM_FAILURE: ();
+    static CREATE_MULTIPART_GENERATION_LOCK_TIMEOUT: ();
 }
 
 #[cfg(test)]
@@ -2429,28 +2430,52 @@ impl S3Access for FS {
 
     /// Checks whether the CreateMultipartUpload request has accesses to the resources.
     async fn create_multipart_upload(&self, req: &mut S3Request<CreateMultipartUploadInput>) -> S3Result<()> {
-        let bucket = req.input.bucket.clone();
-        let bucket_generation = load_bucket_generation(self, req, &bucket).await;
-        let req_info = ext_req_info_mut(&mut req.extensions)?;
-        req_info.bucket = Some(req.input.bucket.clone());
-        req_info.object = Some(req.input.key.clone());
+        let result = async {
+            let bucket = req.input.bucket.clone();
+            let bucket_generation = load_bucket_generation(self, req, &bucket).await;
+            #[cfg(test)]
+            let bucket_generation = if CREATE_MULTIPART_GENERATION_LOCK_TIMEOUT.try_with(|_| ()).is_ok() {
+                Err(ApiError::from(StorageError::Lock(rustfs_lock::LockError::timeout(
+                    format!(".rustfs.sys/bucket-targets/{bucket}/transaction.lock@latest"),
+                    std::time::Duration::from_secs(5),
+                )))
+                .into())
+            } else {
+                bucket_generation
+            };
+            let req_info = ext_req_info_mut(&mut req.extensions)?;
+            req_info.bucket = Some(req.input.bucket.clone());
+            req_info.object = Some(req.input.key.clone());
 
-        let requested_retain_until = req.input.object_lock_retain_until_date.clone();
-        set_requested_object_lock_retain_until(req, requested_retain_until.as_ref());
-        authorize_request(req, Action::S3Action(S3Action::PutObjectAction)).await?;
+            let requested_retain_until = req.input.object_lock_retain_until_date.clone();
+            set_requested_object_lock_retain_until(req, requested_retain_until.as_ref());
+            authorize_request(req, Action::S3Action(S3Action::PutObjectAction)).await?;
 
-        authorize_replication_only_put_headers(req).await?;
+            authorize_replication_only_put_headers(req).await?;
 
-        if legal_hold_write_requested(req.input.object_lock_legal_hold_status.as_ref()) {
-            authorize_request(req, Action::S3Action(S3Action::PutObjectLegalHoldAction)).await?;
+            if legal_hold_write_requested(req.input.object_lock_legal_hold_status.as_ref()) {
+                authorize_request(req, Action::S3Action(S3Action::PutObjectLegalHoldAction)).await?;
+            }
+
+            if retention_write_requested(req.input.object_lock_mode.as_ref(), req.input.object_lock_retain_until_date.as_ref()) {
+                authorize_request(req, Action::S3Action(S3Action::PutObjectRetentionAction)).await?;
+            }
+
+            req.extensions.insert(bucket_generation?);
+            Ok(())
         }
-
-        if retention_write_requested(req.input.object_lock_mode.as_ref(), req.input.object_lock_retain_until_date.as_ref()) {
-            authorize_request(req, Action::S3Action(S3Action::PutObjectRetentionAction)).await?;
+        .await;
+        if let Err(error) = &result {
+            let context = request_context_from_req(req);
+            log_multipart_create_timeout(
+                error,
+                MultipartCreateBoundary::Access,
+                context.as_ref().map(|context| context.request_id.as_str()),
+                &req.input.bucket,
+                &req.input.key,
+            );
         }
-
-        req.extensions.insert(bucket_generation?);
-        Ok(())
+        result
     }
 
     /// Checks whether the DeleteBucket request has accesses to the resources.
@@ -5538,6 +5563,85 @@ mod tests {
         });
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn multipart_timeout_observes_only_the_authorized_final_error() {
+        use crate::error::multipart_timeout_test_log::Capture;
+        use crate::storage::request_context::RequestContext;
+        use tracing::instrument::WithSubscriber as _;
+
+        let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+        let server_ctx = ServerContextSlot::new();
+        assert!(server_ctx.install(Arc::new(AppContext::new(Arc::clone(&store), Arc::new(UnreadyIam), Arc::new(TestKms)))));
+        let fs = FS::with_server_ctx(server_ctx);
+
+        for effect in ["Allow", "Deny"] {
+            let bucket = format!("multipart-timeout-access-{}", uuid::Uuid::new_v4());
+            store
+                .make_bucket(&bucket, &MakeBucketOptions::default())
+                .await
+                .expect("create bucket");
+            let policy_json = format!(
+                r#"{{"Version":"2012-10-17","Statement":[{{"Effect":"{effect}","Principal":{{"AWS":"*"}},"Action":["s3:PutObject"],"Resource":["arn:aws:s3:::{bucket}/*"]}}]}}"#
+            );
+            let mut metadata = (*crate::storage::get_bucket_metadata(&bucket).await.expect("cached metadata")).clone();
+            metadata.policy_config = Some(serde_json::from_str(&policy_json).expect("policy"));
+            metadata.policy_config_json = policy_json.into_bytes();
+            crate::storage::storage_api::set_bucket_metadata(bucket.clone(), metadata)
+                .await
+                .expect("install policy");
+
+            let input = CreateMultipartUploadInput::builder()
+                .bucket(bucket.clone())
+                .key("LEAK-private-key".to_string())
+                .build()
+                .expect("multipart input");
+            let mut req = build_request(input, Method::POST);
+            ensure_req_info(&mut req);
+            req.extensions.insert(fs.server_ctx().clone());
+            let context = RequestContext::from_external_headers(&req.headers);
+            let request_id = context.request_id.clone();
+            req.extensions.insert(context);
+
+            let capture = Capture::default();
+            let subscriber = tracing_subscriber::fmt()
+                .json()
+                .flatten_event(true)
+                .with_current_span(true)
+                .with_span_list(false)
+                .with_max_level(tracing::Level::WARN)
+                .with_writer(capture.clone())
+                .finish();
+            let error = super::CREATE_MULTIPART_GENERATION_LOCK_TIMEOUT
+                .scope((), fs.create_multipart_upload(&mut req))
+                .with_subscriber(subscriber)
+                .await
+                .expect_err("generation timeout must not admit creation");
+            let rows = capture
+                .rows()
+                .into_iter()
+                .filter(|row| row["event"] == "multipart_create_lock_timeout")
+                .collect::<Vec<_>>();
+            if effect == "Allow" {
+                assert_eq!(error.code(), &S3ErrorCode::ServiceUnavailable);
+                assert_eq!(error.headers().unwrap()["x-rustfs-error-cause"], "namespace_lock_timeout");
+                assert_eq!(error.headers().unwrap()[http::header::RETRY_AFTER], "1");
+                assert_eq!(rows.len(), 1);
+                assert_eq!(rows[0]["boundary"], "access");
+                assert_eq!(rows[0]["request_id"], request_id);
+                assert_eq!(rows[0]["resource_class"], "bucket_transaction");
+                let output = rows[0].to_string();
+                assert!(!output.contains(&bucket));
+                assert!(!output.contains("LEAK-private-key"));
+            } else {
+                assert_eq!(error.code(), &S3ErrorCode::AccessDenied);
+                assert_eq!(error.status_code(), Some(http::StatusCode::FORBIDDEN));
+                assert!(rows.is_empty(), "a hidden generation error must not be logged");
+            }
+            assert!(req.extensions.get::<BucketGenerationGuard>().is_none());
+        }
+    }
+
     #[tokio::test]
     #[serial]
     async fn put_read_quorum_access_keeps_authorization_precedence() {

rustfs/src/storage/storage_api.rs

@@ -142,8 +142,8 @@ pub(crate) mod access_consumer {
         authorize_request, bucket_config_mutation_incarnation, delete_object_authorize_action, has_bypass_governance_header,
         load_bucket_generation_from_store, log_list_buckets_iam_implicit_deny, odm_read_generation,
         prepare_list_buckets_iam_authorization, prepare_odm_read_generation, recursive_force_delete_has_authenticated_caller,
-        replication_request_authorized, req_info_mut, req_info_ref, retained_table_data_plane_publication_guards,
-        set_requested_object_lock_retain_until,
+        replication_request_authorized, req_info_mut, req_info_ref, request_context_from_req,
+        retained_table_data_plane_publication_guards, set_requested_object_lock_retain_until,
     };
 }