diff --git a/.github/workflows/package.yml b/.github/workflows/package.yml index d44205fa0..b9536d263 100644 --- a/.github/workflows/package.yml +++ b/.github/workflows/package.yml @@ -15,28 +15,35 @@ # Package Workflow - Build DEB/RPM packages # # This workflow builds DEB and RPM packages from pre-built Linux binaries -# and uploads them to Cloudflare R2. +# and uploads them to Cloudflare R2 and the GitHub release. # # Trigger: -# - release published: automatically package when a GitHub release is published -# - workflow_dispatch: manual trigger with optional tag/run_id +# - workflow_run: automatically package after "Build and Release" completes +# for a release tag (the mac/windows/linux binaries are already uploaded +# to the GitHub release before packaging starts) +# - workflow_dispatch: manual fallback (backfill / re-run) with optional tag/run_id # # Flow: -# 1. Find the Build workflow run for the release tag +# 1. Resolve the triggering Build workflow run for the release tag # 2. Download Linux binaries (x86_64-gnu, aarch64-gnu) from build artifacts # 3. Build DEB packages for amd64 and arm64 # 4. Build RPM packages for x86_64 and aarch64 -# 5. Upload all packages to Cloudflare R2 +# 5. Upload all packages to Cloudflare R2 and the GitHub release name: Package DEB/RPM permissions: - contents: read + # contents: write is required to upload packages to the GitHub release + contents: write actions: read on: - release: - types: [ published ] + # Follows the same pattern as docker.yml: run after the release build + # workflow completes, so packaging is triggered only by release tags + # (e.g. 1.0.0-rc.2, 1.0.0-rc.3), never by development builds. + workflow_run: + workflows: [ "Build and Release" ] + types: [ completed ] workflow_dispatch: inputs: tag: @@ -49,13 +56,26 @@ on: type: string concurrency: - group: ${{ github.workflow }}-${{ github.event.release.tag_name || github.event.inputs.tag || github.run_id }} + group: ${{ github.workflow }}-${{ github.event.workflow_run.head_branch || github.event.inputs.tag || github.run_id }} cancel-in-progress: true +env: + HEAD_BRANCH: ${{ github.event.workflow_run.head_branch }} + WORKFLOW_RUN_ID: ${{ github.event.workflow_run.id }} + jobs: # Resolve which build run to use and extract version info resolve: name: Resolve Build + # Auto-trigger only from successful tag builds of "Build and Release". + # Tag pushes arrive as event == push with head_branch != main (a + # non-main push head_branch is the release tag name). Manual dispatch + # stays available as a fallback for backfills and re-runs. + if: >- + github.event_name == 'workflow_dispatch' || + (github.event.workflow_run.conclusion == 'success' && + github.event.workflow_run.event == 'push' && + github.event.workflow_run.head_branch != 'main') runs-on: ubuntu-latest timeout-minutes: 10 outputs: @@ -75,8 +95,8 @@ jobs: set -euo pipefail # Determine tag - if [[ "${{ github.event_name }}" == "release" ]]; then - TAG="${{ github.event.release.tag_name }}" + if [[ "${{ github.event_name }}" == "workflow_run" ]]; then + TAG="${HEAD_BRANCH}" elif [[ -n "$INPUT_TAG" ]]; then TAG="$INPUT_TAG" else @@ -93,6 +113,11 @@ jobs: BUILD_RUN_ID="$INPUT_RUN_ID" echo "Using explicit build run ID: $BUILD_RUN_ID" + elif [[ "${{ github.event_name }}" == "workflow_run" ]]; then + # Use the Build and Release run that triggered this workflow + BUILD_RUN_ID="${WORKFLOW_RUN_ID}" + echo "Using triggering workflow run: $BUILD_RUN_ID" + elif [[ -n "$TAG" ]]; then # Find the build run that produced this tag echo "Looking for build run for tag: $TAG" @@ -456,6 +481,54 @@ jobs: echo "✅ Latest packages updated" fi + - name: Upload packages to GitHub Release + if: needs.resolve.outputs.tag != '' + env: + GH_TOKEN: ${{ github.token }} + shell: bash + run: | + set -euo pipefail + + TAG="${{ needs.resolve.outputs.tag }}" + DEB_FILE="${{ steps.deb.outputs.deb_file }}" + RPM_FILE="${{ steps.rpm.outputs.rpm_file }}" + + # Upload the packages, then refresh the release checksums so the new + # assets are covered, matching the binary release flow. + for f in "$DEB_FILE" "$RPM_FILE"; do + if [[ -n "$f" && -f "$f" ]]; then + echo "📤 Uploading $(basename "$f") to GitHub release ${TAG}..." + gh release upload "$TAG" "$f" --clobber + fi + done + + CHECKSUM_DIR="$(mktemp -d)" + gh release download "$TAG" -p 'SHA256SUMS' -p 'SHA512SUMS' \ + -D "$CHECKSUM_DIR" --clobber 2>/dev/null || true + + for spec in "SHA256SUMS:sha256sum" "SHA512SUMS:sha512sum"; do + asset="${spec%%:*}" + checksum_cmd="${spec##*:}" + checksum_file="${CHECKSUM_DIR}/${asset}" + + touch "$checksum_file" + + for f in "$DEB_FILE" "$RPM_FILE"; do + if [[ -n "$f" && -f "$f" ]]; then + base="$(basename "$f")" + # Remove any stale entry, then append the fresh digest + grep -Fv -- "$base" "$checksum_file" > "${checksum_file}.tmp" || true + mv "${checksum_file}.tmp" "$checksum_file" + (cd "$(dirname "$f")" && "$checksum_cmd" -- "$base") >> "$checksum_file" + fi + done + + echo "📤 Updating ${asset} for release ${TAG}..." + gh release upload "$TAG" "$checksum_file" --clobber + done + + echo "✅ GitHub release assets updated" + # Summary summary: name: Summary diff --git a/crates/concurrency/src/backpressure.rs b/crates/concurrency/src/backpressure.rs index 502d4fb0f..e692d7329 100644 --- a/crates/concurrency/src/backpressure.rs +++ b/crates/concurrency/src/backpressure.rs @@ -14,9 +14,8 @@ //! Shared backpressure policy type. //! -//! The runtime backpressure implementation (byte-watermark pipes and -//! monitors) lives in `rustfs/src/storage/backpressure.rs`; this module only -//! carries the watermark policy type that implementation shares. +//! This module only carries the watermark policy; the admission primitive it +//! projects into lives in `rustfs-io-core`. use rustfs_io_core::BackpressureConfig as CoreBackpressureConfig; diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index ec5aaf74e..326b798ee 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -1783,7 +1783,7 @@ impl TransitionState { .await; } global_metrics().record_scanner_transition_failed(1); - if !is_err_version_not_found(&err) && !is_err_object_not_found(&err) && !is_network_or_host_down(&err.to_string(), false) && !err.to_string().contains("use of closed network connection") { + if !is_err_version_not_found(&err) && !is_err_object_not_found(&err) && !is_network_or_host_down(&err.to_string(), false) { error!( event = EVENT_LIFECYCLE_TIER_OPERATION_FAILED, component = LOG_COMPONENT_ECSTORE, diff --git a/crates/ecstore/src/client/api_get_options.rs b/crates/ecstore/src/client/api_get_options.rs index 503b44f3a..3438453b3 100644 --- a/crates/ecstore/src/client/api_get_options.rs +++ b/crates/ecstore/src/client/api_get_options.rs @@ -20,6 +20,7 @@ #![allow(clippy::all)] use http::{HeaderMap, HeaderName, HeaderValue}; +use rustfs_utils::http::headers::AMZ_CHECKSUM_MODE; use std::collections::HashMap; use time::OffsetDateTime; use tracing::warn; @@ -76,7 +77,7 @@ impl GetObjectOptions { } } if self.checksum { - headers.insert(HeaderName::from_static("x-amz-checksum-mode"), HeaderValue::from_static("ENABLED")); + headers.insert(HeaderName::from_static(AMZ_CHECKSUM_MODE), HeaderValue::from_static("ENABLED")); } headers } diff --git a/crates/ecstore/src/client/transition_api.rs b/crates/ecstore/src/client/transition_api.rs index 6ad802418..be8e0a510 100644 --- a/crates/ecstore/src/client/transition_api.rs +++ b/crates/ecstore/src/client/transition_api.rs @@ -54,6 +54,10 @@ use rustfs_config::MAX_S3_CLIENT_RESPONSE_SIZE; use rustfs_rio::HashReader; use rustfs_utils::HashAlgorithm; use rustfs_utils::{ + http::headers::{ + AMZ_CHECKSUM_CRC32, AMZ_CHECKSUM_CRC32C, AMZ_CHECKSUM_CRC64NVME, AMZ_CHECKSUM_MODE, AMZ_CHECKSUM_SHA1, + AMZ_CHECKSUM_SHA256, + }, net::get_endpoint_url, retry::{DEFAULT_RETRY_CAP, DEFAULT_RETRY_UNIT, MAX_JITTER, MAX_RETRY, RetryTimer}, }; @@ -1383,12 +1387,12 @@ pub(crate) fn to_object_info_for_provider( }; // Extract checksums - let checksum_crc32 = get_header("x-amz-checksum-crc32"); - let checksum_crc32c = get_header("x-amz-checksum-crc32c"); - let checksum_sha1 = get_header("x-amz-checksum-sha1"); - let checksum_sha256 = get_header("x-amz-checksum-sha256"); - let checksum_crc64nvme = get_header("x-amz-checksum-crc64nvme"); - let checksum_mode = get_header("x-amz-checksum-mode"); + let checksum_crc32 = get_header(AMZ_CHECKSUM_CRC32); + let checksum_crc32c = get_header(AMZ_CHECKSUM_CRC32C); + let checksum_sha1 = get_header(AMZ_CHECKSUM_SHA1); + let checksum_sha256 = get_header(AMZ_CHECKSUM_SHA256); + let checksum_crc64nvme = get_header(AMZ_CHECKSUM_CRC64NVME); + let checksum_mode = get_header(AMZ_CHECKSUM_MODE); // Build and return the ObjectInfo struct Ok(ObjectInfo { diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index 1426fc91f..3bac43266 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -86,6 +86,25 @@ const PEER_REST_RECOVERY_MAX_BACKOFF: Duration = Duration::from_secs(30); const SCANNER_ACTIVITY_MAX_MESSAGE_SIZE: usize = 1024; const REPLICATION_STATS_MAX_MESSAGE_SIZE: usize = 8 * 1024 * 1024; +/// Error for a peer that reported `success = false` without an `error_info` payload. +/// +/// Same shape as `peer_s3_client::peer_failure_without_details`, over `StorageError` +/// instead of `DiskError`. The message names the operation (and the bucket, where the +/// operation has one) and nothing else, for two reasons: +/// +/// - `finalize_result` classifies failures by message substring, so any text matching +/// `message_has_network_needle` would take an answering peer offline and evict its +/// connection over a plain application-level rejection. +/// - Quorum aggregation (`reduce_errs`) buckets `Io` errors by kind plus rendered +/// message, so a per-peer detail such as the peer address would split one shared +/// failure into single-count buckets and downgrade the dominant error. +fn peer_failure_without_details(op: &str, bucket: Option<&str>) -> Error { + match bucket { + Some(bucket) => Error::other(format!("{op}({bucket}): peer returned failure without error details")), + None => Error::other(format!("{op}: peer returned failure without error details")), + } +} + fn decode_bucket_stats_response(response: GetBucketStatsDataResponse) -> Result { if !response.success { return Err(Error::other( @@ -696,7 +715,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("local_storage_info", None)); } let data = response.storage_info; @@ -719,7 +738,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("server_info", None)); } let data = response.server_properties; @@ -742,7 +761,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("get_cpus", None)); } let data = response.cpus; @@ -765,7 +784,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("get_net_info", None)); } let data = response.net_info; @@ -788,7 +807,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("get_partitions", None)); } let data = response.partitions; @@ -811,7 +830,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("get_os_info", None)); } let data = response.os_info; @@ -832,7 +851,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("get_se_linux_info", None)); } let data = response.sys_services; @@ -857,7 +876,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("get_sys_config", None)); } let data = response.sys_config; @@ -882,7 +901,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("get_sys_errors", None)); } let data = response.sys_errors; @@ -907,7 +926,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("get_mem_info", None)); } let data = response.mem_info; @@ -939,7 +958,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("get_metrics", None)); } let data = response.realtime_metrics; @@ -964,7 +983,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("get_live_events", None)); } Ok(PeerLiveEventsBatch { @@ -989,7 +1008,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("get_proc_info", None)); } let data = response.proc_info; @@ -1016,7 +1035,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("start_profiling", None)); } Ok(()) } @@ -1323,7 +1342,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("load_bucket_metadata", Some(bucket))); } Ok(()) } @@ -1346,7 +1365,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("delete_bucket_metadata", Some(bucket))); } Ok(()) } @@ -1369,7 +1388,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("delete_policy", None)); } Ok(()) } @@ -1392,7 +1411,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("load_policy", None)); } Ok(()) } @@ -1417,7 +1436,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("load_policy_mapping", None)); } Ok(()) } @@ -1440,7 +1459,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("delete_user", None)); } Ok(()) } @@ -1463,7 +1482,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("delete_service_account", None)); } Ok(()) } @@ -1487,7 +1506,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("load_user", None)); } Ok(()) } @@ -1510,7 +1529,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("load_service_account", None)); } Ok(()) } @@ -1533,7 +1552,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("load_group", None)); } Ok(()) } @@ -1554,7 +1573,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("reload_site_replication_config", None)); } Ok(()) } @@ -1597,7 +1616,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("signal_service", None)); } validate_signal_service_protocol(sig, sub_sys, response.protocol_version)?; Ok(response) @@ -1667,7 +1686,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("reload_pool_meta", None)); } Ok(()) @@ -1691,7 +1710,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("stop_rebalance", None)); } Ok(()) @@ -1725,7 +1744,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("load_rebalance_meta", None)); } Ok(()) @@ -1753,7 +1772,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("start_decommission", None)); } Ok(()) @@ -1777,7 +1796,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("decommission_cancel", None)); } Ok(()) @@ -1801,7 +1820,7 @@ impl PeerRestClient { if let Some(msg) = response.error_info { return Err(Error::other(msg)); } - return Err(Error::other("")); + return Err(peer_failure_without_details("clear_decommission", None)); } Ok(()) @@ -1947,6 +1966,8 @@ fn tier_config_reload_status_outcome(status: tonic::Status) -> TierConfigReloadO mod tests { use super::*; use crate::config::com::STORAGE_CLASS_SUB_SYS; + use crate::disk::error::DiskError; + use crate::disk::error_reduce::reduce_errs; use crate::layout::{disks_layout::DisksLayout, endpoints::SetupType}; use rustfs_config::{ENV_KUBERNETES_SERVICE_HOST, ENV_LOCAL_ENDPOINT_HOST, ENV_STARTUP_TOPOLOGY_WAIT_MODE}; use serde_json::Value; @@ -3098,4 +3119,115 @@ mod tests { && span.get("request_id").and_then(Value::as_str) == Some("req-peer-rest") })); } + + /// Every operation name passed to `peer_failure_without_details` in this file. + const PEER_FAILURE_OPS: &[&str] = &[ + "local_storage_info", + "server_info", + "get_cpus", + "get_net_info", + "get_partitions", + "get_os_info", + "get_se_linux_info", + "get_sys_config", + "get_sys_errors", + "get_mem_info", + "get_metrics", + "get_live_events", + "get_proc_info", + "start_profiling", + "load_bucket_metadata", + "delete_bucket_metadata", + "delete_policy", + "load_policy", + "load_policy_mapping", + "delete_user", + "delete_service_account", + "load_user", + "load_service_account", + "load_group", + "reload_site_replication_config", + "signal_service", + "reload_pool_meta", + "stop_rebalance", + "load_rebalance_meta", + "start_decommission", + "decommission_cancel", + "clear_decommission", + ]; + + #[test] + fn peer_failure_without_details_names_operation_and_bucket() { + for op in PEER_FAILURE_OPS { + let message = peer_failure_without_details(op, None).to_string(); + assert!(message.contains(op), "{op} message must name the operation: {message}"); + } + + for op in ["load_bucket_metadata", "delete_bucket_metadata"] { + let message = peer_failure_without_details(op, Some("ops-bucket")).to_string(); + assert!(message.contains(op), "{op} message must name the operation: {message}"); + assert!(message.contains("ops-bucket"), "{op} message must name the bucket: {message}"); + } + } + + #[test] + fn peer_failure_without_details_keeps_one_reduce_errs_bucket_per_operation() { + // reduce_errs groups Io errors by kind plus rendered message: peers failing the + // same operation must stay a single dominant error instead of one bucket per peer. + let per_peer_errs = (0..4) + .map(|_| Some(DiskError::from(peer_failure_without_details("load_bucket_metadata", Some("shared"))))) + .collect::>(); + let (count, dominant) = reduce_errs(&per_peer_errs, &[]); + assert_eq!(count, 4, "one shared failure must not split into per-peer buckets"); + assert_eq!( + dominant, + Some(DiskError::from(peer_failure_without_details("load_bucket_metadata", Some("shared")))) + ); + + assert_ne!( + peer_failure_without_details("load_bucket_metadata", Some("shared")).to_string(), + peer_failure_without_details("delete_bucket_metadata", Some("shared")).to_string() + ); + assert_ne!( + peer_failure_without_details("load_bucket_metadata", Some("bucket-a")).to_string(), + peer_failure_without_details("load_bucket_metadata", Some("bucket-b")).to_string() + ); + } + + #[test] + fn peer_failure_without_details_never_reads_as_a_network_failure() { + // `finalize_result` marks the peer offline and evicts its connection whenever the + // message matches a network needle. A peer that answered `success = false` is alive, + // so no operation or bucket name may push this text over that classifier. + for op in PEER_FAILURE_OPS { + let err = peer_failure_without_details(op, None); + assert!( + !PeerRestClient::is_network_like_error(&err), + "{op} must not read as a transport failure: {err}" + ); + + let scoped = peer_failure_without_details(op, Some("bucket-name")); + assert!( + !PeerRestClient::is_network_like_error(&scoped), + "{op} must not read as a transport failure: {scoped}" + ); + } + + // The bucket name is caller-supplied. Every needle carries a space, which S3 bucket + // names cannot, and the name is closed by `)` before the literal text resumes, so no + // needle can straddle the boundary either. + for bucket in [ + "timed-out", + "connection-reset", + "transport-error", + "broken-pipe", + "unavailable-logs", + ] { + let err = peer_failure_without_details("load_bucket_metadata", Some(bucket)); + assert!( + !PeerRestClient::is_network_like_error(&err), + "bucket {bucket} must not push the message over the network classifier: {err}" + ); + } + } } diff --git a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs index 02c9f75f1..d44651dce 100644 --- a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs @@ -220,6 +220,8 @@ fn pool_write_quorum(participant_count: usize) -> usize { /// buckets `Error::Io` by kind plus rendered message, so any per-peer detail (address, /// timing) would split one shared failure into single-count buckets and downgrade a real /// dominant error into `ErasureWriteQuorum`. +/// +/// `peer_rest_client` carries the same helper over `StorageError` for the same response shape. fn peer_failure_without_details(op: &str, bucket: Option<&str>) -> Error { match bucket { Some(bucket) => Error::other(format!("{op}({bucket}): peer returned failure without error details")), diff --git a/crates/ecstore/src/error/mod.rs b/crates/ecstore/src/error/mod.rs index 76dcc7a0c..f8215d908 100644 --- a/crates/ecstore/src/error/mod.rs +++ b/crates/ecstore/src/error/mod.rs @@ -1116,6 +1116,14 @@ mod tests { assert!(encoder_source.is::()); } + // The lifecycle transition worker relies on this arm alone to suppress the + // closed-connection noise (`bucket_lifecycle_ops.rs`); dropping it here would + // silently turn shutdown races back into `error!` log spam. + #[test] + fn is_network_or_host_down_covers_closed_network_connection() { + assert!(is_network_or_host_down("transition failed: use of closed network connection", false)); + } + // Regression for #952 (ECA-11): an all-`DiskNotFound` slice (every drive in // every set unreachable) must NOT be classified as "all not found", // otherwise ListObjects silently returns an empty listing and masks a full diff --git a/docs/architecture/global-state-inventory.md b/docs/architecture/global-state-inventory.md index 8407ad864..b110374dd 100644 --- a/docs/architecture/global-state-inventory.md +++ b/docs/architecture/global-state-inventory.md @@ -105,7 +105,6 @@ inventory. Generic function-local names such as `CACHE`, `LOCK`, `INIT`, and | `USE_STARSHARD_CACHE`, `BUCKET_CACHE_SMALL`, `BUCKET_CACHE_LARGE` | `rustfs/src/storage/ecfs_extend.rs` | Cache or constant / owner-local cache | Bucket validation cache backend selection and cache storage stay private to the ECFS extension owner. | | `GLOBAL_SSE_DEK_PROVIDER`, `SSE_TEST_LOCK` | `rustfs/src/storage/sse.rs` | Owner-local cache / test state | SSE DEK provider cache and test serialization lock stay private to the SSE owner. | | `AUTH_FS` | `rustfs/src/storage/access.rs` | Cache or constant / owner-local cache | Authorization tag-condition lookup keeps its filesystem helper private to the access owner. | -| `LOCK_STATS` | `rustfs/src/storage/lock_optimizer.rs` | Process-global owner-local metrics | Lock optimization statistics stay private behind lock optimizer helper APIs. | | `DEADLOCK_DETECTOR` | `rustfs/src/storage/deadlock_detector.rs` | Process-global owner-local state | Deadlock detector lifecycle state stays private to the storage deadlock detector owner. | | `CONCURRENCY_MANAGER`, `ACTIVE_GET_REQUESTS`, `ACTIVE_PUT_REQUESTS` | `rustfs/src/storage/concurrency/*` | Process-global owner-local scheduler state | Storage concurrency manager and request counters remain inside the storage concurrency owner boundary. | | `GET_OBJECT_BUFFER_THRESHOLD_WARNED`, `GET_READER_STREAM_BUFFER_SIZE_OVERRIDE`, function-local `ENABLED`, `OBJECT_SEEK_SUPPORT_THRESHOLD`, `OBJECT_SEEK_SUPPORT_CONCURRENCY_THRESHOLDS` | `rustfs/src/app/object_usecase.rs` | Cache or constant / owner-local cache | Object GET/seek tuning caches and warning guards stay private to object usecase helpers. | diff --git a/rustfs/src/admin/handlers/kms_dynamic.rs b/rustfs/src/admin/handlers/kms_dynamic.rs index 887ec7fad..d7ee7a0f7 100644 --- a/rustfs/src/admin/handlers/kms_dynamic.rs +++ b/rustfs/src/admin/handlers/kms_dynamic.rs @@ -22,6 +22,7 @@ use crate::admin::runtime_sources::{ current_object_store_handle_for_context, current_or_init_kms_runtime_service_manager, }; use crate::admin::storage_api::config::{read_admin_config, save_admin_config}; +use crate::admin::storage_api::error::StorageError; use crate::auth::{check_key_valid, get_session_token}; use crate::server::{ADMIN_PREFIX, RemoteAddr}; use hyper::{Method, StatusCode}; @@ -278,8 +279,11 @@ pub async fn load_kms_config() -> Option { } }, Err(e) => { - // Config not found is normal on first run - if e.to_string().contains("ConfigNotFound") || e.to_string().contains("not found") { + // Config not found is normal on first run: `read_config` maps a missing or + // empty config object to `ConfigNotFound`, so that variant is the only + // "absent" signal reaching here. Every other not-found variant (disk, + // volume, bucket) means degraded storage and must stay a warning. + if matches!(e, StorageError::ConfigNotFound) { info!( component = LOG_COMPONENT_ADMIN, subsystem = LOG_SUBSYSTEM_KMS, diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index 9eb893792..717de0bee 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -67,6 +67,9 @@ use rustfs_policy::policy::action::{Action, S3Action}; use rustfs_s3_types::EventName; use rustfs_signer::pre_sign_v4; use rustfs_utils::egress::{OutboundDnsResolver, OutboundPolicy}; +use rustfs_utils::http::headers::{ + AMZ_CHECKSUM_CRC32, AMZ_CHECKSUM_CRC32C, AMZ_CHECKSUM_CRC64NVME, AMZ_CHECKSUM_SHA1, AMZ_CHECKSUM_SHA256, AMZ_CHECKSUM_TYPE, +}; use rustfs_utils::http::{ SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK, SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, get_source_scheme, insert_header, @@ -1031,28 +1034,24 @@ fn build_get_object_response_headers(output: &GetObjectOutput, base_headers: &He )?; } if let Some(checksum_crc32) = &output.checksum_crc32 { - insert_string_header(&mut headers, HeaderName::from_static("x-amz-checksum-crc32"), checksum_crc32.clone())?; + insert_string_header(&mut headers, HeaderName::from_static(AMZ_CHECKSUM_CRC32), checksum_crc32.clone())?; } if let Some(checksum_crc32c) = &output.checksum_crc32c { - insert_string_header(&mut headers, HeaderName::from_static("x-amz-checksum-crc32c"), checksum_crc32c.clone())?; + insert_string_header(&mut headers, HeaderName::from_static(AMZ_CHECKSUM_CRC32C), checksum_crc32c.clone())?; } if let Some(checksum_crc64nvme) = &output.checksum_crc64nvme { - insert_string_header( - &mut headers, - HeaderName::from_static("x-amz-checksum-crc64nvme"), - checksum_crc64nvme.clone(), - )?; + insert_string_header(&mut headers, HeaderName::from_static(AMZ_CHECKSUM_CRC64NVME), checksum_crc64nvme.clone())?; } if let Some(checksum_sha1) = &output.checksum_sha1 { - insert_string_header(&mut headers, HeaderName::from_static("x-amz-checksum-sha1"), checksum_sha1.clone())?; + insert_string_header(&mut headers, HeaderName::from_static(AMZ_CHECKSUM_SHA1), checksum_sha1.clone())?; } if let Some(checksum_sha256) = &output.checksum_sha256 { - insert_string_header(&mut headers, HeaderName::from_static("x-amz-checksum-sha256"), checksum_sha256.clone())?; + insert_string_header(&mut headers, HeaderName::from_static(AMZ_CHECKSUM_SHA256), checksum_sha256.clone())?; } if let Some(checksum_type) = &output.checksum_type { insert_string_header( &mut headers, - HeaderName::from_static("x-amz-checksum-type"), + HeaderName::from_static(AMZ_CHECKSUM_TYPE), checksum_type.as_str().to_string(), )?; } @@ -1114,12 +1113,12 @@ fn clear_object_lambda_variant_headers(headers: &mut HeaderMap) { http::header::ETAG, http::header::LAST_MODIFIED, http::header::EXPIRES, - HeaderName::from_static("x-amz-checksum-crc32"), - HeaderName::from_static("x-amz-checksum-crc32c"), - HeaderName::from_static("x-amz-checksum-crc64nvme"), - HeaderName::from_static("x-amz-checksum-sha1"), - HeaderName::from_static("x-amz-checksum-sha256"), - HeaderName::from_static("x-amz-checksum-type"), + HeaderName::from_static(AMZ_CHECKSUM_CRC32), + HeaderName::from_static(AMZ_CHECKSUM_CRC32C), + HeaderName::from_static(AMZ_CHECKSUM_CRC64NVME), + HeaderName::from_static(AMZ_CHECKSUM_SHA1), + HeaderName::from_static(AMZ_CHECKSUM_SHA256), + HeaderName::from_static(AMZ_CHECKSUM_TYPE), HeaderName::from_static("x-amz-tagging-count"), HeaderName::from_static("x-amz-request-route"), HeaderName::from_static("x-amz-request-token"), diff --git a/rustfs/src/storage/backpressure.rs b/rustfs/src/storage/backpressure.rs deleted file mode 100644 index 9585ba8a8..000000000 --- a/rustfs/src/storage/backpressure.rs +++ /dev/null @@ -1,618 +0,0 @@ -// 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. - -//! Backpressure Management for Object Data Transfer. -//! -//! This module provides backpressure-aware pipes for object data transfer, -//! preventing buffer overflow and memory exhaustion under high concurrency. - -//! # Key Features -//! -//! - Configurable buffer size with high/low watermarks -//! - Backpressure state monitoring and events -//! - Backpressure metrics emitted through the shared metrics pipeline -//! - Graceful handling of slow consumers -//! -//! # Architecture -//! -//! ```text -//! [Disk Reader] --> [BackpressurePipe] --> [HTTP Response] -//! | -//! v -//! [Buffer Monitor] -//! | -//! v -//! [High Watermark?] --> Apply Backpressure -//! ``` - -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; -use std::time::{Duration, Instant}; -use tokio::io::{DuplexStream, duplex}; -use tracing::{debug, warn}; - -use metrics::counter; -use rustfs_concurrency::PipeBackpressurePolicy; -use rustfs_io_core::BackpressureConfig as CoreBackpressureConfig; - -/// Object-transfer duplex pipe backpressure policy. -#[derive(Debug, Clone, Copy)] -pub struct ObjectPipeBackpressurePolicy { - /// Buffer size in bytes (default 4MB). - pub buffer_size: usize, - /// High watermark percentage (default 80%). - /// When buffer usage exceeds this, backpressure is applied. - pub high_watermark: u32, - /// Low watermark percentage (default 50%). - /// When buffer usage drops below this after high watermark, backpressure is released. - pub low_watermark: u32, -} - -impl Default for ObjectPipeBackpressurePolicy { - fn default() -> Self { - Self { - buffer_size: rustfs_config::DEFAULT_OBJECT_DUPLEX_BUFFER_SIZE, - high_watermark: rustfs_config::DEFAULT_OBJECT_BACKPRESSURE_HIGH_WATERMARK, - low_watermark: rustfs_config::DEFAULT_OBJECT_BACKPRESSURE_LOW_WATERMARK, - } - } -} - -impl ObjectPipeBackpressurePolicy { - /// Load configuration from environment variables. - pub fn from_env() -> Self { - let buffer_size = rustfs_utils::get_env_usize( - rustfs_config::ENV_OBJECT_DUPLEX_BUFFER_SIZE, - rustfs_config::DEFAULT_OBJECT_DUPLEX_BUFFER_SIZE, - ); - let high_watermark = rustfs_utils::get_env_u32( - rustfs_config::ENV_OBJECT_BACKPRESSURE_HIGH_WATERMARK, - rustfs_config::DEFAULT_OBJECT_BACKPRESSURE_HIGH_WATERMARK, - ); - let low_watermark = rustfs_utils::get_env_u32( - rustfs_config::ENV_OBJECT_BACKPRESSURE_LOW_WATERMARK, - rustfs_config::DEFAULT_OBJECT_BACKPRESSURE_LOW_WATERMARK, - ); - - Self { - buffer_size, - high_watermark, - low_watermark, - } - } - - /// Calculate high watermark threshold in bytes. - pub fn high_watermark_bytes(&self) -> usize { - (self.buffer_size as u64 * self.high_watermark as u64 / 100) as usize - } - - /// Calculate low watermark threshold in bytes. - pub fn low_watermark_bytes(&self) -> usize { - (self.buffer_size as u64 * self.low_watermark as u64 / 100) as usize - } - - /// Project this object-transfer policy into the shared concurrency facade policy. - pub fn to_concurrency_policy(&self) -> PipeBackpressurePolicy { - PipeBackpressurePolicy { - buffer_size: self.buffer_size, - high_watermark: self.high_watermark, - low_watermark: self.low_watermark, - } - } - - /// Project this object-transfer policy into the reusable io-core admission config. - pub fn to_core_config(&self) -> CoreBackpressureConfig { - self.to_concurrency_policy().to_core_config() - } -} - -/// Backpressure state. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum BackpressureState { - /// Normal operation, buffer usage is below high watermark. - Normal, - /// Buffer usage is above high watermark, backpressure should be applied. - HighWatermark, - /// Backpressure is actively being applied to the producer. - BackpressureApplied, -} - -impl std::fmt::Display for BackpressureState { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - BackpressureState::Normal => write!(f, "normal"), - BackpressureState::HighWatermark => write!(f, "high_watermark"), - BackpressureState::BackpressureApplied => write!(f, "backpressure_applied"), - } - } -} - -/// Compact metadata snapshot for object-transfer backpressure pipes. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct BackpressurePipeMeta { - /// Buffer capacity in bytes. - pub buffer_capacity: usize, - /// Current backpressure state. - pub state: BackpressureState, - /// Age of the pipe since creation. - pub age: Duration, -} - -/// Compact metadata snapshot for the lightweight backpressure monitor. -#[derive(Debug, Clone, Copy, PartialEq)] -pub struct BackpressureMonitorMeta { - /// Buffer capacity in bytes. - pub buffer_capacity: usize, - /// Current buffer usage percentage. - pub usage_percent: f32, - /// Current backpressure state. - pub state: BackpressureState, -} - -fn calculate_usage_percent(usage: usize, capacity: usize) -> f32 { - if capacity > 0 { - (usage as f32 / capacity as f32) * 100.0 - } else { - 0.0 - } -} - -fn apply_watermark_transition( - in_high_watermark: &AtomicBool, - usage: usize, - high: usize, - low: usize, -) -> (BackpressureState, bool) { - let current = in_high_watermark.load(Ordering::Acquire); - let next_state = if usage >= high { - BackpressureState::HighWatermark - } else if usage <= low { - BackpressureState::Normal - } else if current { - BackpressureState::HighWatermark - } else { - BackpressureState::Normal - }; - let next_is_high = matches!(next_state, BackpressureState::HighWatermark); - let changed = in_high_watermark.swap(next_is_high, Ordering::AcqRel) != next_is_high; - (next_state, changed) -} - -fn saturating_sub_atomic(value: &AtomicUsize, delta: usize) { - value - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| Some(current.saturating_sub(delta))) - .ok(); -} - -/// A backpressure-aware pipe wrapping tokio's duplex. -/// -/// This provides monitoring and events for backpressure conditions -/// while maintaining compatibility with the standard duplex interface. -pub struct BackpressurePipe { - /// Reader end of the duplex pipe. - reader: DuplexStream, - /// Writer end of the duplex pipe. - writer: DuplexStream, - /// Configuration. - config: ObjectPipeBackpressurePolicy, - /// Current buffer usage (approximate, updated on write). - buffer_usage: AtomicUsize, - /// Current backpressure state. - state: AtomicBool, // true = in high watermark state - /// Total bytes written. - total_written: AtomicUsize, - /// Total bytes read. - total_read: AtomicUsize, - /// Cached high watermark threshold in bytes. - high_watermark_bytes: usize, - /// Cached low watermark threshold in bytes. - low_watermark_bytes: usize, - /// Pipe creation timestamp. - created_at: Instant, -} - -impl BackpressurePipe { - /// Create a new backpressure-aware pipe with default configuration. - pub fn new() -> Self { - Self::with_config(ObjectPipeBackpressurePolicy::from_env()) - } - - /// Create a new backpressure-aware pipe with custom configuration. - pub fn with_config(config: ObjectPipeBackpressurePolicy) -> Self { - let policy = config.to_concurrency_policy(); - let (reader, writer) = duplex(policy.buffer_size); - let high_watermark_bytes = policy.high_watermark_bytes(); - let low_watermark_bytes = policy.low_watermark_bytes(); - - debug!( - buffer_size = config.buffer_size, - high_watermark = config.high_watermark, - low_watermark = config.low_watermark, - high_watermark_bytes, - low_watermark_bytes, - "Created backpressure pipe" - ); - - Self { - reader, - writer, - config, - buffer_usage: AtomicUsize::new(0), - state: AtomicBool::new(false), - total_written: AtomicUsize::new(0), - total_read: AtomicUsize::new(0), - high_watermark_bytes, - low_watermark_bytes, - created_at: Instant::now(), - } - } - - /// Take the reader end of the pipe (consumes self). - pub fn into_reader(self) -> DuplexStream { - self.reader - } - - /// Take the writer end of the pipe (consumes self). - pub fn into_writer(self) -> DuplexStream { - self.writer - } - - /// Split into reader and writer (consumes self). - pub fn split(self) -> (DuplexStream, DuplexStream) { - (self.reader, self.writer) - } - - /// Get current backpressure state. - pub fn state(&self) -> BackpressureState { - if self.state.load(Ordering::Acquire) { - BackpressureState::BackpressureApplied - } else { - BackpressureState::Normal - } - } - - /// Get a compact metadata snapshot for the pipe. - pub fn meta(&self) -> BackpressurePipeMeta { - BackpressurePipeMeta { - buffer_capacity: self.config.buffer_size, - state: self.state(), - age: self.age(), - } - } - - /// Get the age of this pipe. - pub fn age(&self) -> Duration { - self.created_at.elapsed() - } - - /// Get current buffer usage. - pub fn usage(&self) -> usize { - self.buffer_usage.load(Ordering::Acquire) - } - - /// Record bytes written (call after successful write). - pub fn record_write(&self, bytes: usize) { - self.total_written.fetch_add(bytes, Ordering::Relaxed); - self.buffer_usage.fetch_add(bytes, Ordering::Release); - self.update_watermark_state(); - } - - /// Record bytes read (call after successful read). - pub fn record_read(&self, bytes: usize) { - self.total_read.fetch_add(bytes, Ordering::Relaxed); - saturating_sub_atomic(&self.buffer_usage, bytes); - self.update_watermark_state(); - } - - /// Update watermark state and emit transition signals. - fn update_watermark_state(&self) { - let usage = self.buffer_usage.load(Ordering::Acquire); - let usage_percent = calculate_usage_percent(usage, self.config.buffer_size) as u32; - let (next_state, changed) = - apply_watermark_transition(&self.state, usage, self.high_watermark_bytes, self.low_watermark_bytes); - - if changed { - match next_state { - BackpressureState::HighWatermark => { - counter!("rustfs_backpressure_events_total", "state" => "high_watermark").increment(1); - - warn!( - buffer_usage = usage, - buffer_capacity = self.config.buffer_size, - usage_percent, - high_watermark = self.config.high_watermark, - "Backpressure: high watermark reached" - ); - } - BackpressureState::Normal => { - counter!("rustfs_backpressure_events_total", "state" => "normal").increment(1); - - debug!( - buffer_usage = usage, - buffer_capacity = self.config.buffer_size, - usage_percent, - low_watermark = self.config.low_watermark, - "Backpressure: returned to normal" - ); - } - BackpressureState::BackpressureApplied => {} - } - } - } - - /// Get total bytes written. - pub fn total_written(&self) -> usize { - self.total_written.load(Ordering::Relaxed) - } - - /// Get total bytes read. - pub fn total_read(&self) -> usize { - self.total_read.load(Ordering::Relaxed) - } - - /// Get buffer capacity. - pub fn capacity(&self) -> usize { - self.config.buffer_size - } -} - -impl Default for BackpressurePipe { - fn default() -> Self { - Self::new() - } -} - -/// A simple wrapper that provides backpressure monitoring for duplex streams. -/// -/// This is a lighter-weight alternative to `BackpressurePipe` that doesn't -/// wrap the streams but provides monitoring capabilities. -pub struct BackpressureMonitor { - /// Configuration. - config: ObjectPipeBackpressurePolicy, - /// Current buffer usage. - buffer_usage: AtomicUsize, - /// In high watermark state. - in_high_watermark: AtomicBool, - /// Cached high watermark threshold in bytes. - high_watermark_bytes: usize, - /// Cached low watermark threshold in bytes. - low_watermark_bytes: usize, -} - -impl BackpressureMonitor { - /// Create a new monitor with default configuration. - pub fn new() -> Self { - Self::with_config(ObjectPipeBackpressurePolicy::from_env()) - } - - /// Create a new monitor with custom configuration. - pub fn with_config(config: ObjectPipeBackpressurePolicy) -> Self { - let policy = config.to_concurrency_policy(); - let high_watermark_bytes = policy.high_watermark_bytes(); - let low_watermark_bytes = policy.low_watermark_bytes(); - Self { - config, - buffer_usage: AtomicUsize::new(0), - in_high_watermark: AtomicBool::new(false), - high_watermark_bytes, - low_watermark_bytes, - } - } - - /// Record bytes added to buffer. - pub fn on_write(&self, bytes: usize) -> BackpressureState { - self.buffer_usage.fetch_add(bytes, Ordering::Release); - self.update_state() - } - - /// Record bytes removed from buffer. - pub fn on_read(&self, bytes: usize) -> BackpressureState { - saturating_sub_atomic(&self.buffer_usage, bytes); - self.update_state() - } - - /// Get current state. - pub fn state(&self) -> BackpressureState { - if self.in_high_watermark.load(Ordering::Acquire) { - BackpressureState::HighWatermark - } else { - BackpressureState::Normal - } - } - - /// Get current buffer usage. - pub fn usage(&self) -> usize { - self.buffer_usage.load(Ordering::Acquire) - } - - /// Get usage percentage. - pub fn usage_percent(&self) -> f32 { - let usage = self.buffer_usage.load(Ordering::Acquire); - calculate_usage_percent(usage, self.config.buffer_size) - } - - /// Get a compact metadata snapshot for the monitor. - pub fn meta(&self) -> BackpressureMonitorMeta { - let usage = self.buffer_usage.load(Ordering::Acquire); - BackpressureMonitorMeta { - buffer_capacity: self.config.buffer_size, - usage_percent: calculate_usage_percent(usage, self.config.buffer_size), - state: self.state(), - } - } - - /// Update state based on current usage. - fn update_state(&self) -> BackpressureState { - let usage = self.buffer_usage.load(Ordering::Acquire); - let usage_percent = calculate_usage_percent(usage, self.config.buffer_size) as u32; - let (next_state, changed) = - apply_watermark_transition(&self.in_high_watermark, usage, self.high_watermark_bytes, self.low_watermark_bytes); - - if matches!(next_state, BackpressureState::HighWatermark) { - if changed { - counter!("rustfs_backpressure_events_total", "state" => "high_watermark").increment(1); - - debug!(usage_percent, "Backpressure: entered high watermark"); - } - BackpressureState::HighWatermark - } else { - if changed { - counter!("rustfs_backpressure_events_total", "state" => "normal").increment(1); - - debug!(usage_percent, "Backpressure: returned to normal"); - } - BackpressureState::Normal - } - } -} - -impl Default for BackpressureMonitor { - fn default() -> Self { - Self::new() - } -} - -#[cfg(test)] -#[allow(unused_imports)] -mod tests { - use super::{BackpressureMonitor, BackpressurePipe, BackpressureState, ObjectPipeBackpressurePolicy}; - - #[test] - fn test_backpressure_config_default() { - let config = ObjectPipeBackpressurePolicy::default(); - assert_eq!(config.buffer_size, 4 * 1024 * 1024); - assert_eq!(config.high_watermark, 80); - assert_eq!(config.low_watermark, 50); - } - - #[test] - fn test_backpressure_config_watermarks() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 1000, - high_watermark: 80, - low_watermark: 50, - }; - assert_eq!(config.high_watermark_bytes(), 800); - assert_eq!(config.low_watermark_bytes(), 500); - } - - #[test] - fn test_backpressure_policy_projects_to_concurrency_and_core_config() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 2000, - high_watermark: 75, - low_watermark: 40, - }; - let concurrency = config.to_concurrency_policy(); - let core = config.to_core_config(); - - assert_eq!(concurrency.buffer_size, config.buffer_size); - assert_eq!(concurrency.high_watermark, config.high_watermark); - assert_eq!(concurrency.low_watermark, config.low_watermark); - assert_eq!(core.high_water_mark, 0.75); - assert_eq!(core.low_water_mark, 0.40); - assert!(core.enabled); - } - - #[test] - fn test_backpressure_pipe_consumes_concurrency_policy_thresholds() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 2000, - high_watermark: 75, - low_watermark: 40, - }; - let concurrency = config.to_concurrency_policy(); - let pipe = BackpressurePipe::with_config(config); - - assert_eq!(pipe.capacity(), concurrency.buffer_size); - assert_eq!(pipe.high_watermark_bytes, concurrency.high_watermark_bytes()); - assert_eq!(pipe.low_watermark_bytes, concurrency.low_watermark_bytes()); - } - - #[test] - fn test_backpressure_monitor_consumes_concurrency_policy_thresholds() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 2000, - high_watermark: 75, - low_watermark: 40, - }; - let concurrency = config.to_concurrency_policy(); - let monitor = BackpressureMonitor::with_config(config); - - assert_eq!(monitor.meta().buffer_capacity, concurrency.buffer_size); - assert_eq!(monitor.high_watermark_bytes, concurrency.high_watermark_bytes()); - assert_eq!(monitor.low_watermark_bytes, concurrency.low_watermark_bytes()); - } - - #[test] - fn test_backpressure_state_display() { - assert_eq!(format!("{}", BackpressureState::Normal), "normal"); - assert_eq!(format!("{}", BackpressureState::HighWatermark), "high_watermark"); - assert_eq!(format!("{}", BackpressureState::BackpressureApplied), "backpressure_applied"); - } - - #[test] - fn test_backpressure_monitor() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 1000, - high_watermark: 80, - low_watermark: 50, - }; - let monitor = BackpressureMonitor::with_config(config); - - // Initially normal - assert_eq!(monitor.state(), BackpressureState::Normal); - assert_eq!(monitor.meta().buffer_capacity, 1000); - assert_eq!(monitor.meta().usage_percent, 0.0); - - // Write to reach high watermark - let state = monitor.on_write(850); - assert_eq!(state, BackpressureState::HighWatermark); - assert_eq!(monitor.meta().usage_percent, 85.0); - - // Read to go below low watermark - let state = monitor.on_read(400); - assert_eq!(state, BackpressureState::Normal); - assert_eq!(monitor.meta().usage_percent, 45.0); - } - - #[tokio::test] - async fn test_backpressure_pipe_creation() { - let pipe = BackpressurePipe::new(); - assert_eq!(pipe.capacity(), 4 * 1024 * 1024); - assert_eq!(pipe.state(), BackpressureState::Normal); - assert_eq!(pipe.meta().buffer_capacity, 4 * 1024 * 1024); - assert!(pipe.meta().age <= pipe.age()); - } - - #[test] - fn test_backpressure_pipe_state_transitions() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 1000, - high_watermark: 80, - low_watermark: 50, - }; - let pipe = BackpressurePipe::with_config(config); - - assert_eq!(pipe.state(), BackpressureState::Normal); - assert_eq!(pipe.meta().state, BackpressureState::Normal); - - pipe.record_write(850); - assert_eq!(pipe.state(), BackpressureState::BackpressureApplied); - assert_eq!(pipe.meta().state, BackpressureState::BackpressureApplied); - - pipe.record_read(400); - assert_eq!(pipe.state(), BackpressureState::Normal); - assert_eq!(pipe.meta().state, BackpressureState::Normal); - } -} diff --git a/rustfs/src/storage/concurrent_fix_test.rs b/rustfs/src/storage/concurrent_fix_test.rs index 2fedeced1..a80865976 100644 --- a/rustfs/src/storage/concurrent_fix_test.rs +++ b/rustfs/src/storage/concurrent_fix_test.rs @@ -14,17 +14,15 @@ //! Integration tests for concurrent request fix. //! -//! These tests verify that the timeout, backpressure, and deadlock detection -//! mechanisms work correctly under high concurrency scenarios. +//! These tests verify that the timeout and deadlock detection mechanisms work +//! correctly under high concurrency scenarios. #[cfg(test)] mod tests { - use crate::storage::backpressure::{BackpressureMonitor, BackpressureState, ObjectPipeBackpressurePolicy}; use crate::storage::concurrency::{IoLoadLevel, IoPriority}; use crate::storage::deadlock_detector::{ DeadlockDetector, LockInfo, LockType, RequestHangDetectionPolicy, RequestResourceTracker, }; - use crate::storage::lock_optimizer::{LockOptimizeConfig, LockOptimizer, LockStats}; use crate::storage::timeout_wrapper::{GetObjectTimeoutPolicy, RequestTimeoutWrapper, TimedGetObjectResult}; use std::time::Duration; @@ -114,82 +112,6 @@ mod tests { } } - // ============================================ - // Backpressure Tests - // ============================================ - - #[test] - fn test_backpressure_config_defaults() { - let config = ObjectPipeBackpressurePolicy::default(); - assert_eq!(config.buffer_size, 4 * 1024 * 1024); // 4MB - assert_eq!(config.high_watermark, 80); - assert_eq!(config.low_watermark, 50); - } - - #[test] - fn test_backpressure_monitor_state_transitions() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 1000, - high_watermark: 80, - low_watermark: 50, - }; - let monitor = BackpressureMonitor::with_config(config); - - // Initially normal - assert_eq!(monitor.state(), BackpressureState::Normal); - - // Write to reach high watermark - let state = monitor.on_write(850); - assert_eq!(state, BackpressureState::HighWatermark); - - // Read to go below low watermark - let state = monitor.on_read(400); - assert_eq!(state, BackpressureState::Normal); - } - - #[test] - fn test_backpressure_usage_percent() { - let config = ObjectPipeBackpressurePolicy { - buffer_size: 1000, - high_watermark: 80, - low_watermark: 50, - }; - let monitor = BackpressureMonitor::with_config(config); - - monitor.on_write(500); - assert!((monitor.usage_percent() - 50.0).abs() < 1.0); - } - - // ============================================ - // Lock Optimizer Tests - // ============================================ - - #[test] - fn test_lock_optimize_config_defaults() { - let config = LockOptimizeConfig::default(); - assert!(config.enabled); - assert_eq!(config.acquire_timeout, Duration::from_secs(5)); - } - - #[test] - fn test_lock_stats_tracking() { - let stats = LockStats::new(); - - stats.record_acquire(); - stats.record_early_release(Duration::from_millis(100)); - stats.record_early_release(Duration::from_millis(200)); - - assert_eq!(stats.locks_acquired.load(std::sync::atomic::Ordering::Relaxed), 1); - assert_eq!(stats.locks_released_early.load(std::sync::atomic::Ordering::Relaxed), 2); - assert_eq!(stats.max_hold_time(), Duration::from_millis(200)); - } - - #[test] - fn test_lock_optimizer_creation() { - let optimizer = LockOptimizer::new(); - assert!(optimizer.is_enabled()); - } - // ============================================ // I/O Priority Tests // ============================================ diff --git a/rustfs/src/storage/lock_optimizer.rs b/rustfs/src/storage/lock_optimizer.rs deleted file mode 100644 index 3636c50c3..000000000 --- a/rustfs/src/storage/lock_optimizer.rs +++ /dev/null @@ -1,458 +0,0 @@ -// 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. - -//! Lock Optimization for GetObject Operations. -//! -//! This module provides optimized lock management for read operations, -//! reducing lock contention by releasing locks early (after metadata read) -//! rather than holding them for the entire data transfer duration. -//! -//! # Migration Note -//! -//! For new code, consider using `rustfs_io_core::LockOptimizer` which provides -//! the same core functionality with better separation of concerns. This module -//! remains for backward compatibility and storage-specific configuration. -//! -//! ```ignore -//! // Recommended: Use io-core directly -//! use rustfs_io_core::LockOptimizer; -//! let optimizer = LockOptimizer::with_defaults(); -//! ``` - -// Allow dead_code for public API that may be used by external modules or future features -//! # Key Features -//! -//! - Early lock release after metadata read -//! - Lock hold time monitoring -//! - Configurable optimization (can be disabled for debugging) -//! - Lock contention metrics emitted through the shared metrics pipeline -//! -//! # Architecture -//! -//! ```text -//! Traditional: [Acquire Lock] --> [Read Metadata] --> [Transfer Data] --> [Release Lock] -//! |<------------------ Lock Held ------------------>| -//! -//! Optimized: [Acquire Lock] --> [Read Metadata] --> [Release Lock] --> [Transfer Data] -//! |<- Lock Held ->| -//! ``` - -use std::sync::Arc; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::time::{Duration, Instant}; -use tracing::debug; - -use metrics::histogram; - -/// Lock optimization configuration. -#[derive(Debug, Clone)] -pub struct LockOptimizeConfig { - /// Whether to enable lock optimization. - /// When enabled, read locks are released after metadata read. - /// When disabled, locks are held for the entire operation (traditional behavior). - pub enabled: bool, - /// Lock acquisition timeout. - pub acquire_timeout: Duration, -} - -impl Default for LockOptimizeConfig { - fn default() -> Self { - Self { - enabled: rustfs_config::DEFAULT_OBJECT_LOCK_OPTIMIZATION_ENABLE, - acquire_timeout: Duration::from_secs(rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT), - } - } -} - -impl LockOptimizeConfig { - /// Load configuration from environment variables. - pub fn from_env() -> Self { - let enabled = rustfs_utils::get_env_bool( - rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, - rustfs_config::DEFAULT_OBJECT_LOCK_OPTIMIZATION_ENABLE, - ); - let acquire_timeout = Duration::from_secs(rustfs_utils::get_env_u64( - rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, - rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT, - )); - - Self { - enabled, - acquire_timeout, - } - } -} - -/// Statistics for lock optimization monitoring. -#[derive(Debug, Default)] -pub struct LockStats { - /// Total locks acquired. - pub locks_acquired: AtomicU64, - /// Total locks released early. - pub locks_released_early: AtomicU64, - /// Total lock hold time in microseconds. - pub total_hold_time_us: AtomicU64, - /// Maximum lock hold time in microseconds. - pub max_hold_time_us: AtomicU64, -} - -impl LockStats { - /// Create new lock statistics. - pub fn new() -> Self { - Self::default() - } - - /// Record a lock acquisition. - pub fn record_acquire(&self) { - self.locks_acquired.fetch_add(1, Ordering::Relaxed); - } - - /// Record an early lock release. - pub fn record_early_release(&self, hold_time: Duration) { - self.locks_released_early.fetch_add(1, Ordering::Relaxed); - self.record_hold_time(hold_time); - } - - /// Record lock hold time. - fn record_hold_time(&self, hold_time: Duration) { - let hold_time_us = hold_time.as_micros() as u64; - self.total_hold_time_us.fetch_add(hold_time_us, Ordering::Relaxed); - - // Update max hold time - let mut current_max = self.max_hold_time_us.load(Ordering::Relaxed); - while hold_time_us > current_max { - match self - .max_hold_time_us - .compare_exchange_weak(current_max, hold_time_us, Ordering::Relaxed, Ordering::Relaxed) - { - Ok(_) => break, - Err(actual) => current_max = actual, - } - } - } - - /// Get average hold time. - pub fn avg_hold_time(&self) -> Duration { - let total = self.total_hold_time_us.load(Ordering::Relaxed); - let count = self.locks_released_early.load(Ordering::Relaxed); - total.checked_div(count).map(Duration::from_micros).unwrap_or(Duration::ZERO) - } - - /// Get maximum hold time. - pub fn max_hold_time(&self) -> Duration { - Duration::from_micros(self.max_hold_time_us.load(Ordering::Relaxed)) - } -} - -/// Global lock statistics. -static LOCK_STATS: std::sync::OnceLock> = std::sync::OnceLock::new(); - -/// Get global lock statistics. -pub fn get_lock_stats() -> Arc { - LOCK_STATS.get_or_init(|| Arc::new(LockStats::new())).clone() -} - -/// An optimized lock guard that supports early release. -/// -/// This wraps the actual lock guard and provides: -/// - Early release capability (before drop) -/// - Hold time tracking -/// - Metrics reporting -pub struct OptimizedLockGuard { - /// The underlying lock guard. - guard: Option, - /// When the lock was acquired. - acquire_time: Instant, - /// Whether the lock has been released. - released: bool, - /// Lock resource name (for logging). - resource: String, - /// Statistics reference. - stats: Arc, -} - -impl OptimizedLockGuard { - /// Create a new optimized lock guard. - pub fn new(guard: G, resource: impl Into) -> Self { - let stats = get_lock_stats(); - stats.record_acquire(); - - Self { - guard: Some(guard), - acquire_time: Instant::now(), - released: false, - resource: resource.into(), - stats, - } - } - - /// Get the lock hold time so far. - pub fn hold_time(&self) -> Duration { - self.acquire_time.elapsed() - } - - /// Check if the lock has been released. - pub fn is_released(&self) -> bool { - self.released - } - - /// Release the lock early (before drop). - /// - /// This is the key optimization: releasing the lock after - /// metadata read rather than waiting for the entire operation. - pub fn early_release(&mut self) { - if self.released { - return; - } - - let hold_time = self.hold_time(); - self.guard.take(); - self.released = true; - - self.stats.record_early_release(hold_time); - - histogram!("rustfs_lock_hold_duration_seconds").record(hold_time.as_secs_f64()); - - debug!( - resource = %self.resource, - hold_time_ms = hold_time.as_millis(), - "Lock released early (optimization active)" - ); - } - - /// Get a reference to the underlying guard. - pub fn as_ref(&self) -> Option<&G> { - if self.released { None } else { self.guard.as_ref() } - } -} - -impl Drop for OptimizedLockGuard { - fn drop(&mut self) { - if !self.released { - let hold_time = self.hold_time(); - self.guard.take(); - self.released = true; - - self.stats.record_early_release(hold_time); - - histogram!("rustfs_lock_hold_duration_seconds").record(hold_time.as_secs_f64()); - - debug!( - resource = %self.resource, - hold_time_ms = hold_time.as_millis(), - "Lock released on drop (normal release)" - ); - } - } -} - -/// A scope guard that releases a lock when it goes out of scope. -/// -/// This is a simpler version of OptimizedLockGuard for cases -/// where we just need RAII semantics without tracking. -pub struct LockScopeGuard { - guard: Option, -} - -impl LockScopeGuard { - /// Create a new scope guard. - pub fn new(guard: G) -> Self { - Self { guard: Some(guard) } - } - - /// Release the lock early. - pub fn release(&mut self) { - self.guard.take(); - } -} - -impl Drop for LockScopeGuard { - fn drop(&mut self) { - self.guard.take(); - } -} - -/// Helper for managing lock optimization in GetObject operations. -/// -/// This provides a clean interface for the common pattern: -/// 1. Acquire lock -/// 2. Read metadata -/// 3. Release lock (if optimization enabled) -/// 4. Transfer data (without lock) -pub struct LockOptimizer { - /// Configuration. - config: LockOptimizeConfig, -} - -impl LockOptimizer { - /// Create a new lock optimizer with default configuration. - pub fn new() -> Self { - Self { - config: LockOptimizeConfig::from_env(), - } - } - - /// Create a new lock optimizer with custom configuration. - pub fn with_config(config: LockOptimizeConfig) -> Self { - Self { config } - } - - /// Check if lock optimization is enabled. - pub fn is_enabled(&self) -> bool { - self.config.enabled - } - - /// Get the lock acquisition timeout. - pub fn acquire_timeout(&self) -> Duration { - self.config.acquire_timeout - } - - /// Wrap a lock guard for optimization. - pub fn wrap_guard(&self, guard: G, resource: impl Into) -> OptimizedLockGuard { - OptimizedLockGuard::new(guard, resource) - } - - /// Execute a metadata read operation with lock optimization. - /// - /// This is the main entry point for optimized lock usage: - /// - If optimization is enabled: lock is released after metadata_fn completes - /// - If optimization is disabled: lock is held until the returned guard is dropped - /// - /// # Arguments - /// - /// * `guard` - The lock guard to optimize - /// * `resource` - Resource name for logging - /// * `metadata_fn` - Function to read metadata while holding lock - /// - /// # Returns - /// - /// A tuple of (metadata result, optional guard to hold for later release) - pub async fn with_optimized_lock( - &self, - guard: G, - resource: impl Into, - metadata_fn: F, - ) -> (T, Option>) - where - F: FnOnce() -> Fut, - Fut: std::future::Future, - { - let resource = resource.into(); - let mut optimized = OptimizedLockGuard::new(guard, &resource); - - // Execute metadata read while holding lock - let result = metadata_fn().await; - - if self.config.enabled { - // Release lock early - optimized.early_release(); - (result, None) - } else { - // Keep lock for caller to release - (result, Some(optimized)) - } - } -} - -impl Default for LockOptimizer { - fn default() -> Self { - Self::new() - } -} - -/// Check if lock optimization is enabled globally. -pub fn is_lock_optimization_enabled() -> bool { - rustfs_utils::get_env_bool( - rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, - rustfs_config::DEFAULT_OBJECT_LOCK_OPTIMIZATION_ENABLE, - ) -} - -#[cfg(test)] -#[allow(unused_imports)] -mod tests { - use super::{LockOptimizeConfig, LockOptimizer, LockStats, OptimizedLockGuard}; - use std::sync::Mutex; - use std::sync::atomic::Ordering; - use std::time::Duration; - - #[test] - fn test_lock_optimize_config_default() { - let config = LockOptimizeConfig::default(); - assert!(config.enabled); - assert_eq!(config.acquire_timeout, Duration::from_secs(5)); - } - - #[test] - fn test_lock_stats() { - let stats = LockStats::new(); - - stats.record_acquire(); - stats.record_early_release(Duration::from_millis(100)); - stats.record_early_release(Duration::from_millis(200)); - - assert_eq!(stats.locks_acquired.load(Ordering::Relaxed), 1); - assert_eq!(stats.locks_released_early.load(Ordering::Relaxed), 2); - assert_eq!(stats.max_hold_time(), Duration::from_millis(200)); - } - - #[test] - fn test_optimized_lock_guard() { - let guard = Mutex::new(42); - let locked = guard.lock().unwrap(); - let mut optimized = OptimizedLockGuard::new(locked, "test-resource"); - - assert!(!optimized.is_released()); - assert!(optimized.hold_time() < Duration::from_secs(1)); - - optimized.early_release(); - assert!(optimized.is_released()); - } - - #[test] - fn test_lock_optimizer() { - let optimizer = LockOptimizer::new(); - assert!(optimizer.is_enabled()); - } - - #[tokio::test] - async fn test_with_optimized_lock_enabled() { - let optimizer = LockOptimizer::new(); - let guard = Mutex::new(42); - let locked = guard.lock().unwrap(); - - let (result, returned_guard) = optimizer.with_optimized_lock(locked, "test-resource", || async { 100 }).await; - - assert_eq!(result, 100); - // With optimization enabled, guard should be None (released early) - assert!(returned_guard.is_none()); - } - - #[tokio::test] - async fn test_with_optimized_lock_disabled() { - let config = LockOptimizeConfig { - enabled: false, - acquire_timeout: Duration::from_secs(5), - }; - let optimizer = LockOptimizer::with_config(config); - let guard = Mutex::new(42); - let locked = guard.lock().unwrap(); - - let (result, returned_guard) = optimizer.with_optimized_lock(locked, "test-resource", || async { 100 }).await; - - assert_eq!(result, 100); - // With optimization disabled, guard should be Some (held for later) - assert!(returned_guard.is_some()); - } -} diff --git a/rustfs/src/storage/mod.rs b/rustfs/src/storage/mod.rs index 58b6a0ce9..e0817da90 100644 --- a/rustfs/src/storage/mod.rs +++ b/rustfs/src/storage/mod.rs @@ -13,12 +13,10 @@ // limitations under the License. pub mod access; -pub mod backpressure; pub mod concurrency; pub mod deadlock_detector; pub mod ecfs; pub(crate) mod helper; -pub mod lock_optimizer; pub mod options; pub mod request_context; pub mod rpc;