Compare commits

..

4 Commits

Author SHA1 Message Date
cxymds 24127ed230 Merge branch 'main' into cxymds/fix-1852-remote-recovery 2026-08-14 21:01:56 +08:00
马登山 334323bd6f test(ecstore): cover remote recovery review cases 2026-08-14 18:19:35 +08:00
cxymds d35c8e1066 Merge branch 'main' into cxymds/fix-1852-remote-recovery 2026-08-14 13:53:27 +08:00
马登山 595c563cc1 fix(ecstore): single-flight remote disk recovery 2026-08-14 10:34:10 +08:00
254 changed files with 5269 additions and 23842 deletions
+1 -7
View File
@@ -252,16 +252,10 @@ test-group = 'ecstore-serial-flaky'
# cluster, so it keeps the lane's parallel-safe / no-external-dependency
# properties. The RustFS warm backend has no loopback guard (that guard is
# replication-only), so it needs no opt-in env for its 127.0.0.1 tier target.
#
# Disk compression (backlog#1848): the `compression` module joins the smoke
# lane so the multipart disk-compression roundtrips (restored after
# rustfs/rustfs#5169 disabled them) have PR-lane signal, not just merge-gate.
# Single-node servers on random ports with isolated temp dirs — meets the
# admission criteria unchanged.
[profile.e2e-smoke]
default-filter = """
package(e2e_test) & (
test(/^(delete_marker_migration_semantics|version_id_regression|list_objects_v2_pagination|list_object_versions_regression|list_objects_duplicates|list_buckets_double_slash|list_buckets_auth|list_buckets_iam_filter|leading_slash_key|special_chars|create_bucket_region|delete_objects_versioning|head_object_consistency|head_object_range|copy_object_metadata|copy_object_tagging|copy_source_invalid_date|content_encoding|compression|multipart_storage_class|storage_class_capability|ssec_copy|anonymous_access|bucket_policy_check|presigned_negative|negative_sigv4|admin_auth|notification_webhook|tls_hot_reload|console_smoke|admin_iam_crud|admin_pools|sts_query_compat)_test::|^fake_s3_target::/)
test(/^(delete_marker_migration_semantics|version_id_regression|list_objects_v2_pagination|list_object_versions_regression|list_objects_duplicates|list_buckets_double_slash|list_buckets_auth|list_buckets_iam_filter|leading_slash_key|special_chars|create_bucket_region|delete_objects_versioning|head_object_consistency|head_object_range|copy_object_metadata|copy_object_tagging|copy_source_invalid_date|content_encoding|multipart_storage_class|storage_class_capability|ssec_copy|anonymous_access|bucket_policy_check|presigned_negative|negative_sigv4|admin_auth|notification_webhook|tls_hot_reload|console_smoke|admin_iam_crud|admin_pools|sts_query_compat)_test::|^fake_s3_target::/)
| test(/^replication_extension_test::(test_replication_check_succeeds_with_remote_target|test_replication_check_rejects_target_without_object_lock|test_set_remote_target_rejects_unversioned_source_bucket|test_replication_check_rejects_unversioned_source_bucket|test_replication_check_rejects_missing_replication_config|test_replication_check_rejects_invalid_bucket|test_set_remote_target_rejects_same_bucket_on_same_deployment|test_set_remote_target_rejects_unversioned_target_bucket|test_set_remote_target_update_requires_arn|test_set_remote_target_update_rejects_missing_target|test_set_remote_target_rejects_invalid_target_url|test_set_remote_target_rejects_self_signed_https_target_without_skip_tls_verify|test_set_remote_target_rejects_private_ca_https_target_without_ca_cert_pem|test_list_remote_targets_rejects_empty_bucket|test_list_remote_targets_rejects_invalid_bucket|test_remove_remote_target_rejects_missing_target|test_remove_remote_target_rejects_missing_arn|test_remove_remote_target_rejects_invalid_bucket|test_remove_remote_target_rejects_target_used_by_replication|test_delete_bucket_replication_removes_remote_target)$/)
| test(/^reliant::lifecycle::/)
| test(/^reliant::tiering::/)
+1 -25
View File
@@ -94,7 +94,6 @@ jobs:
short_sha: ${{ steps.check.outputs.short_sha }}
is_prerelease: ${{ steps.check.outputs.is_prerelease }}
create_latest: ${{ steps.check.outputs.create_latest }}
source_ref: ${{ steps.check.outputs.source_ref }}
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
@@ -119,7 +118,6 @@ jobs:
short_sha=""
is_prerelease=false
create_latest=false
source_ref="$GITHUB_SHA"
if [[ "${{ github.event_name }}" == "workflow_run" ]]; then
# Triggered by build workflow completion
@@ -139,7 +137,6 @@ jobs:
# Extract version info from commit message or use commit SHA
# Use Git to generate consistent short SHA (ensures uniqueness like build.yml)
short_sha=$(git rev-parse --short "$HEAD_SHA")
source_ref="$HEAD_SHA"
# Determine build type based on triggering workflow event and ref
triggering_event="$TRIGGERING_EVENT"
@@ -264,23 +261,6 @@ jobs:
echo "⚠️ Only release versions (latest, v1.0.0, 1.0.0) and prereleases (v1.0.0-alpha1, 1.0.0-beta2) are supported"
;;
esac
if [[ "$should_build" == true && "$input_version" != "latest" ]]; then
tag_ref="refs/tags/$input_version"
if ! git ls-remote --exit-code origin "$tag_ref" >/dev/null 2>&1; then
if [[ "$input_version" == v* ]]; then
tag_ref="refs/tags/${input_version#v}"
else
tag_ref="refs/tags/v$input_version"
fi
fi
if ! git ls-remote --exit-code origin "$tag_ref" >/dev/null 2>&1; then
echo "❌ Release tag not found for Docker build: $input_version"
exit 1
fi
source_ref="$tag_ref"
fi
fi
{
@@ -291,7 +271,6 @@ jobs:
echo "short_sha=$short_sha"
echo "is_prerelease=$is_prerelease"
echo "create_latest=$create_latest"
echo "source_ref=$source_ref"
} >> "$GITHUB_OUTPUT"
echo "🐳 Docker Build Summary:"
@@ -302,7 +281,6 @@ jobs:
echo " - Short SHA: $short_sha"
echo " - Is prerelease: $is_prerelease"
echo " - Create latest: $create_latest"
echo " - Source ref: $source_ref"
# Build multi-arch Docker images
# Strategy: Build images using pre-built binaries from dl.rustfs.com
@@ -330,7 +308,6 @@ jobs:
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
ref: ${{ needs.build-check.outputs.source_ref }}
- name: Login to Docker Hub
uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3
@@ -420,8 +397,7 @@ jobs:
LABELS="org.opencontainers.image.title=RustFS"
LABELS="$LABELS,org.opencontainers.image.description=RustFS distributed object storage system"
LABELS="$LABELS,org.opencontainers.image.version=$VERSION"
SOURCE_REVISION="$(git rev-parse HEAD)"
LABELS="$LABELS,org.opencontainers.image.revision=$SOURCE_REVISION"
LABELS="$LABELS,org.opencontainers.image.revision=${{ github.sha }}"
LABELS="$LABELS,org.opencontainers.image.source=${{ github.server_url }}/${{ github.repository }}"
LABELS="$LABELS,org.opencontainers.image.created=$(date -u +'%Y-%m-%dT%H:%M:%SZ')"
LABELS="$LABELS,org.opencontainers.image.build-type=$BUILD_TYPE"
Generated
+48 -56
View File
@@ -278,7 +278,6 @@ checksum = "312c1ea69e5fe9966e0029fb95aca8790100b85aff4f0d3b00a9337c74069a9c"
dependencies = [
"bigdecimal",
"bon",
"crc32fast",
"digest 0.11.3",
"log",
"miniz_oxide 0.9.1",
@@ -290,11 +289,9 @@ dependencies = [
"serde",
"serde_bytes",
"serde_json",
"snap",
"strum",
"thiserror 2.0.20",
"uuid",
"zstd",
]
[[package]]
@@ -3764,7 +3761,7 @@ checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"
[[package]]
name = "e2e_test"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"anyhow",
"astral-tokio-tar",
@@ -9093,7 +9090,7 @@ dependencies = [
[[package]]
name = "rustfs"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"aes-gcm",
"anyhow",
@@ -9203,7 +9200,6 @@ dependencies = [
"serial_test",
"sha2 0.11.0",
"shadow-rs",
"snap",
"socket2",
"subtle",
"sysinfo",
@@ -9231,7 +9227,7 @@ dependencies = [
[[package]]
name = "rustfs-audit"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"async-trait",
"const-str",
@@ -9254,7 +9250,7 @@ dependencies = [
[[package]]
name = "rustfs-checksums"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"base64-simd",
"bytes",
@@ -9270,7 +9266,7 @@ dependencies = [
[[package]]
name = "rustfs-common"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"chrono",
"hotpath",
@@ -9288,7 +9284,7 @@ dependencies = [
[[package]]
name = "rustfs-concurrency"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"hotpath",
"insta",
@@ -9301,7 +9297,7 @@ dependencies = [
[[package]]
name = "rustfs-config"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"const-str",
"hotpath",
@@ -9311,7 +9307,7 @@ dependencies = [
[[package]]
name = "rustfs-credentials"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"base64-simd",
"hmac 0.13.0",
@@ -9325,7 +9321,7 @@ dependencies = [
[[package]]
name = "rustfs-crypto"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"aes-gcm",
"argon2",
@@ -9346,7 +9342,7 @@ dependencies = [
[[package]]
name = "rustfs-data-usage"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"hotpath",
"rmp-serde",
@@ -9356,7 +9352,7 @@ dependencies = [
[[package]]
name = "rustfs-ecstore"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"arc-swap",
"async-channel",
@@ -9463,6 +9459,7 @@ dependencies = [
"tokio-stream",
"tokio-util",
"tonic",
"tonic-prost",
"tower",
"tracing",
"tracing-core",
@@ -9495,7 +9492,7 @@ dependencies = [
[[package]]
name = "rustfs-extension-schema"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"hotpath",
"serde",
@@ -9505,7 +9502,7 @@ dependencies = [
[[package]]
name = "rustfs-filemeta"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"arc-swap",
"byteorder",
@@ -9532,7 +9529,7 @@ dependencies = [
[[package]]
name = "rustfs-heal"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"async-trait",
"base64 0.23.1",
@@ -9563,7 +9560,7 @@ dependencies = [
[[package]]
name = "rustfs-iam"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"arc-swap",
"async-trait",
@@ -9604,7 +9601,7 @@ dependencies = [
[[package]]
name = "rustfs-io-core"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"bytes",
"hotpath",
@@ -9617,7 +9614,7 @@ dependencies = [
[[package]]
name = "rustfs-io-metrics"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"criterion",
"hotpath",
@@ -9681,7 +9678,7 @@ dependencies = [
[[package]]
name = "rustfs-keystone"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"bytes",
"futures",
@@ -9708,7 +9705,7 @@ dependencies = [
[[package]]
name = "rustfs-kms"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"aes-gcm",
"anyhow",
@@ -9757,7 +9754,7 @@ dependencies = [
[[package]]
name = "rustfs-lifecycle"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"async-trait",
"hotpath",
@@ -9780,7 +9777,7 @@ dependencies = [
[[package]]
name = "rustfs-lock"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"async-trait",
"compact_str",
@@ -9803,7 +9800,7 @@ dependencies = [
[[package]]
name = "rustfs-log-analyzer"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"chrono",
"flate2",
@@ -9822,27 +9819,22 @@ dependencies = [
[[package]]
name = "rustfs-madmin"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"hotpath",
"http 1.5.0",
"humantime",
"hyper",
"jiff",
"reqwest",
"rmp-serde",
"rustfs-signer",
"s3s",
"serde",
"serde_json",
"sysinfo",
"time",
"tokio",
]
[[package]]
name = "rustfs-notify"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"arc-swap",
"async-trait",
@@ -9877,7 +9869,7 @@ dependencies = [
[[package]]
name = "rustfs-object-capacity"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"criterion",
"futures",
@@ -9897,7 +9889,7 @@ dependencies = [
[[package]]
name = "rustfs-object-data-cache"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"bytes",
"criterion",
@@ -9914,7 +9906,7 @@ dependencies = [
[[package]]
name = "rustfs-obs"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"chrono",
"crossbeam-channel",
@@ -9969,7 +9961,7 @@ dependencies = [
[[package]]
name = "rustfs-policy"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"async-trait",
"base64-simd",
@@ -10000,7 +9992,7 @@ dependencies = [
[[package]]
name = "rustfs-protocols"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"astral-tokio-tar",
"async-compression",
@@ -10062,7 +10054,7 @@ dependencies = [
[[package]]
name = "rustfs-protos"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"flatbuffers",
"hotpath",
@@ -10086,7 +10078,7 @@ dependencies = [
[[package]]
name = "rustfs-replication"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"byteorder",
"bytes",
@@ -10104,7 +10096,7 @@ dependencies = [
[[package]]
name = "rustfs-rio"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"aes-gcm",
"arc-swap",
@@ -10142,7 +10134,7 @@ dependencies = [
[[package]]
name = "rustfs-rio-v2"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"aes-gcm",
"bytes",
@@ -10165,7 +10157,7 @@ dependencies = [
[[package]]
name = "rustfs-s3-ops"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"hotpath",
"rustfs-s3-types",
@@ -10173,7 +10165,7 @@ dependencies = [
[[package]]
name = "rustfs-s3-types"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"hotpath",
"serde",
@@ -10182,7 +10174,7 @@ dependencies = [
[[package]]
name = "rustfs-s3select-api"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"async-trait",
"bytes",
@@ -10212,7 +10204,7 @@ dependencies = [
[[package]]
name = "rustfs-s3select-query"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"async-recursion",
"async-trait",
@@ -10231,7 +10223,7 @@ dependencies = [
[[package]]
name = "rustfs-scanner"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"async-trait",
"bytes",
@@ -10271,7 +10263,7 @@ dependencies = [
[[package]]
name = "rustfs-security-governance"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"hotpath",
"thiserror 2.0.20",
@@ -10279,7 +10271,7 @@ dependencies = [
[[package]]
name = "rustfs-signer"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"base64-simd",
"bytes",
@@ -10297,7 +10289,7 @@ dependencies = [
[[package]]
name = "rustfs-storage-api"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"async-trait",
"hotpath",
@@ -10312,7 +10304,7 @@ dependencies = [
[[package]]
name = "rustfs-targets"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"arc-swap",
"async-nats",
@@ -10366,7 +10358,7 @@ dependencies = [
[[package]]
name = "rustfs-test-utils"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"hotpath",
"rustfs-data-usage",
@@ -10382,7 +10374,7 @@ dependencies = [
[[package]]
name = "rustfs-tls-runtime"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"arc-swap",
"hotpath",
@@ -10403,7 +10395,7 @@ dependencies = [
[[package]]
name = "rustfs-trusted-proxies"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"async-trait",
"axum",
@@ -10440,7 +10432,7 @@ dependencies = [
[[package]]
name = "rustfs-utils"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"base64-simd",
"blake2",
@@ -10482,7 +10474,7 @@ dependencies = [
[[package]]
name = "rustfs-zip"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
dependencies = [
"astral-tokio-tar",
"async-compression",
+48 -48
View File
@@ -69,7 +69,7 @@ edition = "2024"
license = "Apache-2.0"
repository = "https://github.com/rustfs/rustfs"
rust-version = "1.97.1"
version = "1.0.0-rc.2"
version = "1.0.0-rc.1"
homepage = "https://rustfs.com"
description = "RustFS is a high-performance distributed object storage software built using Rust, one of the most popular languages worldwide. "
keywords = ["RustFS", "Minio", "object-storage", "filesystem", "s3"]
@@ -86,52 +86,52 @@ redundant_clone = "warn"
[workspace.dependencies]
# RustFS Internal Crates
rustfs = { path = "./rustfs", version = "1.0.0-rc.2" }
rustfs-heal = { path = "crates/heal", version = "1.0.0-rc.2" }
rustfs-audit = { path = "crates/audit", version = "1.0.0-rc.2" }
rustfs-checksums = { path = "crates/checksums", version = "1.0.0-rc.2" }
rustfs-common = { path = "crates/common", version = "1.0.0-rc.2" }
rustfs-data-usage = { path = "crates/data-usage", version = "1.0.0-rc.2" }
rustfs-config = { path = "./crates/config", version = "1.0.0-rc.2" }
rustfs-concurrency = { path = "./crates/concurrency", version = "1.0.0-rc.2" }
rustfs-credentials = { path = "crates/credentials", version = "1.0.0-rc.2" }
rustfs-crypto = { path = "crates/crypto", version = "1.0.0-rc.2" }
rustfs-ecstore = { path = "crates/ecstore", version = "1.0.0-rc.2" }
rustfs-filemeta = { path = "crates/filemeta", version = "1.0.0-rc.2" }
rustfs-iam = { path = "crates/iam", version = "1.0.0-rc.2" }
rustfs-keystone = { path = "crates/keystone", version = "1.0.0-rc.2" }
rustfs-lifecycle = { path = "crates/lifecycle", version = "1.0.0-rc.2" }
rustfs-kms = { path = "crates/kms", version = "1.0.0-rc.2" }
rustfs-lock = { path = "crates/lock", version = "1.0.0-rc.2" }
rustfs-madmin = { path = "crates/madmin", version = "1.0.0-rc.2" }
rustfs-notify = { path = "crates/notify", version = "1.0.0-rc.2" }
rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.0-rc.2" }
rustfs-io-core = { path = "crates/io-core", version = "1.0.0-rc.2" }
rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.0-rc.2" }
rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.0-rc.2", default-features = false }
rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.0-rc.2" }
rustfs-obs = { path = "crates/obs", version = "1.0.0-rc.2" }
rustfs-policy = { path = "crates/policy", version = "1.0.0-rc.2" }
rustfs-protos = { path = "crates/protos", version = "1.0.0-rc.2" }
rustfs-protocols = { path = "crates/protocols", version = "1.0.0-rc.2" }
rustfs-replication = { path = "crates/replication", version = "1.0.0-rc.2" }
rustfs-rio = { path = "crates/rio", version = "1.0.0-rc.2" }
rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.0-rc.2" }
rustfs-s3-types = { path = "crates/s3-types", version = "1.0.0-rc.2" }
rustfs-s3-ops = { path = "crates/s3-ops", version = "1.0.0-rc.2" }
rustfs-s3select-api = { path = "crates/s3select-api", version = "1.0.0-rc.2" }
rustfs-s3select-query = { path = "crates/s3select-query", version = "1.0.0-rc.2" }
rustfs-scanner = { path = "crates/scanner", version = "1.0.0-rc.2" }
rustfs-security-governance = { path = "crates/security-governance", version = "1.0.0-rc.2" }
rustfs-extension-schema = { path = "crates/extension-schema", version = "1.0.0-rc.2" }
rustfs-signer = { path = "crates/signer", version = "1.0.0-rc.2" }
rustfs-storage-api = { path = "crates/storage-api", version = "1.0.0-rc.2" }
rustfs-trusted-proxies = { path = "crates/trusted-proxies", version = "1.0.0-rc.2" }
rustfs-targets = { path = "crates/targets", version = "1.0.0-rc.2" }
rustfs-test-utils = { path = "crates/test-utils", version = "1.0.0-rc.2" }
rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.0-rc.2" }
rustfs-utils = { path = "crates/utils", version = "1.0.0-rc.2" }
rustfs-zip = { path = "./crates/zip", version = "1.0.0-rc.2" }
rustfs = { path = "./rustfs", version = "1.0.0-rc.1" }
rustfs-heal = { path = "crates/heal", version = "1.0.0-rc.1" }
rustfs-audit = { path = "crates/audit", version = "1.0.0-rc.1" }
rustfs-checksums = { path = "crates/checksums", version = "1.0.0-rc.1" }
rustfs-common = { path = "crates/common", version = "1.0.0-rc.1" }
rustfs-data-usage = { path = "crates/data-usage", version = "1.0.0-rc.1" }
rustfs-config = { path = "./crates/config", version = "1.0.0-rc.1" }
rustfs-concurrency = { path = "./crates/concurrency", version = "1.0.0-rc.1" }
rustfs-credentials = { path = "crates/credentials", version = "1.0.0-rc.1" }
rustfs-crypto = { path = "crates/crypto", version = "1.0.0-rc.1" }
rustfs-ecstore = { path = "crates/ecstore", version = "1.0.0-rc.1" }
rustfs-filemeta = { path = "crates/filemeta", version = "1.0.0-rc.1" }
rustfs-iam = { path = "crates/iam", version = "1.0.0-rc.1" }
rustfs-keystone = { path = "crates/keystone", version = "1.0.0-rc.1" }
rustfs-lifecycle = { path = "crates/lifecycle", version = "1.0.0-rc.1" }
rustfs-kms = { path = "crates/kms", version = "1.0.0-rc.1" }
rustfs-lock = { path = "crates/lock", version = "1.0.0-rc.1" }
rustfs-madmin = { path = "crates/madmin", version = "1.0.0-rc.1" }
rustfs-notify = { path = "crates/notify", version = "1.0.0-rc.1" }
rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.0-rc.1" }
rustfs-io-core = { path = "crates/io-core", version = "1.0.0-rc.1" }
rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.0-rc.1" }
rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.0-rc.1", default-features = false }
rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.0-rc.1" }
rustfs-obs = { path = "crates/obs", version = "1.0.0-rc.1" }
rustfs-policy = { path = "crates/policy", version = "1.0.0-rc.1" }
rustfs-protos = { path = "crates/protos", version = "1.0.0-rc.1" }
rustfs-protocols = { path = "crates/protocols", version = "1.0.0-rc.1" }
rustfs-replication = { path = "crates/replication", version = "1.0.0-rc.1" }
rustfs-rio = { path = "crates/rio", version = "1.0.0-rc.1" }
rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.0-rc.1" }
rustfs-s3-types = { path = "crates/s3-types", version = "1.0.0-rc.1" }
rustfs-s3-ops = { path = "crates/s3-ops", version = "1.0.0-rc.1" }
rustfs-s3select-api = { path = "crates/s3select-api", version = "1.0.0-rc.1" }
rustfs-s3select-query = { path = "crates/s3select-query", version = "1.0.0-rc.1" }
rustfs-scanner = { path = "crates/scanner", version = "1.0.0-rc.1" }
rustfs-security-governance = { path = "crates/security-governance", version = "1.0.0-rc.1" }
rustfs-extension-schema = { path = "crates/extension-schema", version = "1.0.0-rc.1" }
rustfs-signer = { path = "crates/signer", version = "1.0.0-rc.1" }
rustfs-storage-api = { path = "crates/storage-api", version = "1.0.0-rc.1" }
rustfs-trusted-proxies = { path = "crates/trusted-proxies", version = "1.0.0-rc.1" }
rustfs-targets = { path = "crates/targets", version = "1.0.0-rc.1" }
rustfs-test-utils = { path = "crates/test-utils", version = "1.0.0-rc.1" }
rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.0-rc.1" }
rustfs-utils = { path = "crates/utils", version = "1.0.0-rc.1" }
rustfs-zip = { path = "./crates/zip", version = "1.0.0-rc.1" }
# Async Runtime and Networking
async-channel = "2.5.0"
@@ -171,7 +171,7 @@ tower = { version = "0.5.3" }
tower-http = { version = "0.7.0" }
# Serialization and Data Formats
apache-avro = { version = "0.22.0", features = ["snappy", "zstandard"] }
apache-avro = "0.22.0"
bytes = { version = "1.12.1" }
bytesize = "2.7.0"
byteorder = "1.5.0"
+1 -1
View File
@@ -116,7 +116,7 @@ chown -R 10001:10001 data logs
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:latest
# Using specific version
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-rc.2
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-rc.1
```
If you use [podman](https://github.com/containers/podman) instead of docker, you can install the RustFS with the below command
+1 -1
View File
@@ -113,7 +113,7 @@ chown -R 10001:10001 data logs
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:latest
# 使用指定版本运行
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-rc.2
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-rc.1
```
如果您通过绑定挂载启用 TLS 证书目录,也请用同样方式准备该目录:
-1
View File
@@ -40,7 +40,6 @@ mak = "mak"
gae = "gae"
GAE = "GAE"
thr = "thr"
mis = "mis"
# s3-tests original test names (cannot be changed)
nonexisted = "nonexisted"
consts = "consts"
+1
View File
@@ -11,6 +11,7 @@
// 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.
#![allow(dead_code)]
use base64_simd::STANDARD;
+1 -4
View File
@@ -38,10 +38,7 @@ pub const XXHASH_3_HEADER_NAME: &str = "x-amz-checksum-xxhash3";
pub const XXHASH_64_HEADER_NAME: &str = "x-amz-checksum-xxhash64";
pub const XXHASH_128_HEADER_NAME: &str = "x-amz-checksum-xxhash128";
#[allow(
dead_code,
reason = "Content-MD5 wire name, resolved by header_name() below and asserted by this crate's tests (backlog#1823)"
)]
#[allow(dead_code)]
pub(crate) static MD5_HEADER_NAME: &str = "content-md5";
pub const CHECKSUM_ALGORITHMS_IN_PRIORITY_ORDER: [&str; 5] =
+2 -8
View File
@@ -476,19 +476,13 @@ impl Checksum for Xxhash64 {
}
}
#[allow(dead_code)]
#[derive(Debug, Default)]
#[allow(
dead_code,
reason = "Content-MD5 is not a ChecksumAlgorithm variant and has no arm in into_impl: S3 carries it as its own header, separate from the x-amz-checksum-* family. This impl exists so the two paths share the Checksum trait, and is asserted by this crate's tests (backlog#1823)"
)]
struct Md5 {
hasher: md5::Md5,
}
#[allow(
dead_code,
reason = "Content-MD5 is not a ChecksumAlgorithm variant and has no arm in into_impl: S3 carries it as its own header, separate from the x-amz-checksum-* family. This impl exists so the two paths share the Checksum trait, and is asserted by this crate's tests (backlog#1823)"
)]
#[allow(dead_code)]
impl Md5 {
fn update(&mut self, bytes: &[u8]) {
use md5::Digest;
-286
View File
@@ -870,157 +870,6 @@ pub struct DataUsageCacheInfo {
pub snapshot_complete: bool,
}
/// Prefix-level usage over a raw entry map — the shared core behind
/// [`DataUsageCache::prefix_usage`], usable by any cache-shaped reader (the
/// scanner's writer-side cache has the same map type).
///
/// Cache keys are cleaned literal paths (`bucket/pre/fix`), so sub-prefix
/// names come straight off the child keys — no reverse mapping exists or is
/// needed. A compacted prefix carries its aggregate but no children, which
/// the `compacted` flag reports so callers can say why the breakdown is
/// empty. `truncated` is set when the breakdown exceeded `max_entries` and
/// was cut (largest first).
pub fn prefix_usage_in_cache(
cache: &HashMap<String, DataUsageEntry>,
bucket: &str,
prefix: &str,
max_entries: usize,
) -> Option<PrefixUsageQuery> {
let prefix = prefix.trim_matches('/');
let root = if prefix.is_empty() {
bucket.to_string()
} else {
format!("{bucket}/{prefix}")
};
let entry = cache.get(&hash_path(&root).key())?.clone();
let usage = PrefixUsageSummary::from_entry(&flatten_entry(cache, &entry, 0)?);
let child_prefix = format!("{root}/");
let mut sub_prefixes: Vec<PrefixUsageEntry> = entry
.children
.iter()
.filter_map(|child_key| {
let child = cache.get(child_key)?;
let child_flat = flatten_entry(cache, child, 1)?;
// Child keys are literal `bucket/pre/name` paths; a trailing
// slash marks a directory object and is display-only here.
let name = child_key
.strip_prefix(child_prefix.as_str())
.unwrap_or(child_key.as_str())
.trim_end_matches('/')
.to_string();
Some(PrefixUsageEntry {
prefix: name,
usage: PrefixUsageSummary::from_entry(&child_flat),
})
})
.collect();
sub_prefixes.sort_by(|left, right| {
right
.usage
.size
.cmp(&left.usage.size)
.then_with(|| left.prefix.cmp(&right.prefix))
});
let truncated = sub_prefixes.len() > max_entries;
sub_prefixes.truncate(max_entries);
Some(PrefixUsageQuery {
usage,
compacted: entry.compacted,
truncated,
sub_prefixes,
})
}
/// Maximum subtree depth [`flatten_entry`] will walk before declaring the
/// cache corrupt — the same bound the scanner's checked flatten uses.
const PREFIX_USAGE_MAX_DEPTH: usize = 1024;
/// Flatten one entry's subtree into an aggregate: the free-function twin of
/// [`DataUsageCache::flatten`], carrying the scanner checked-flatten
/// hardening so a corrupt cache (cycles, over-deep trees, overflowing
/// counters) yields `None` instead of unbounded recursion or wrapped totals.
fn flatten_entry(cache: &HashMap<String, DataUsageEntry>, root: &DataUsageEntry, depth: usize) -> Option<DataUsageEntry> {
if depth > PREFIX_USAGE_MAX_DEPTH {
return None;
}
let mut flattened = DataUsageEntry::default();
if !flattened.checked_merge(root) {
return None;
}
flattened.compacted = root.compacted;
// The root itself is not pre-seeded: it is merged above, and a corrupt
// child edge pointing back at the root's own key is still terminated by
// the visited set on first encounter.
let mut visited: HashSet<&str> = HashSet::new();
let mut pending: Vec<(&String, usize)> = root.children.iter().map(|child| (child, depth + 1)).collect();
while let Some((key, child_depth)) = pending.pop() {
if child_depth > PREFIX_USAGE_MAX_DEPTH || !visited.insert(key.as_str()) {
return None;
}
let entry = cache.get(key)?;
if !flattened.checked_merge(entry) {
return None;
}
pending.extend(entry.children.iter().map(|child| (child, child_depth + 1)));
}
flattened.children.clear();
Some(flattened)
}
/// Flattened counters of one prefix subtree, as returned by
/// [`DataUsageCache::prefix_usage`].
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct PrefixUsageSummary {
pub size: u64,
pub objects: u64,
pub versions: u64,
pub delete_markers: u64,
}
impl PrefixUsageSummary {
fn from_entry(entry: &DataUsageEntry) -> Self {
Self {
size: entry.size as u64,
objects: entry.objects as u64,
versions: entry.versions as u64,
delete_markers: entry.delete_markers as u64,
}
}
/// Add another set's counters into this one (entries are partitioned by
/// set, so per-set results sum).
pub fn merge(&mut self, other: &Self) {
self.size = self.size.saturating_add(other.size);
self.objects = self.objects.saturating_add(other.objects);
self.versions = self.versions.saturating_add(other.versions);
self.delete_markers = self.delete_markers.saturating_add(other.delete_markers);
}
}
/// One first-level sub-prefix row of a [`PrefixUsageQuery`].
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
pub struct PrefixUsageEntry {
pub prefix: String,
pub usage: PrefixUsageSummary,
}
/// Result of [`DataUsageCache::prefix_usage`].
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct PrefixUsageQuery {
pub usage: PrefixUsageSummary,
/// The prefix entry was compacted by the scanner: its aggregate is valid
/// but no sub-prefix breakdown exists on disk.
pub compacted: bool,
/// The breakdown had more entries than `max_entries`; the largest remain.
pub truncated: bool,
pub sub_prefixes: Vec<PrefixUsageEntry>,
}
/// Read-only projection of a scanner-written `.usage-cache.bin` file.
///
/// The scanner-side `DataUsageCache` (`crates/scanner/src/data_usage_define.rs`)
@@ -1148,21 +997,6 @@ impl DataUsageCache {
}
}
/// Prefix-level usage for one bucket subtree, plus the one-level
/// breakdown below it (rustfs/backlog#1872, MinIO
/// `loadPrefixUsageFromBackend` parity and beyond: arbitrary prefixes and
/// full counters instead of first-level sizes only).
///
/// Cache keys are cleaned literal paths (`bucket/pre/fix`), so sub-prefix
/// names come straight off the child keys — no reverse mapping exists or
/// is needed. A compacted prefix carries its aggregate but no children,
/// which the `compacted` flag reports so callers can say why the
/// breakdown is empty. `truncated` is set when the breakdown exceeded
/// `max_entries` and was cut (largest first).
pub fn prefix_usage(&self, bucket: &str, prefix: &str, max_entries: usize) -> Option<PrefixUsageQuery> {
prefix_usage_in_cache(&self.cache, bucket, prefix, max_entries)
}
pub fn force_compact(&mut self, limit: usize) {
if self.cache.len() < limit {
return;
@@ -2064,126 +1898,6 @@ mod tests {
);
}
/// Build a cache shaped like `bucket/{a,b/{c,d}},bucket/loose` with
/// distinct counters so aggregation is observable.
fn prefix_usage_fixture_cache() -> DataUsageCache {
let mut cache = DataUsageCache::default();
let mut insert = |path: &str, parent: &str, size: usize, objects: usize, versions: usize, delete_markers: usize| {
cache.replace(
path,
parent,
DataUsageEntry {
size,
objects,
versions,
delete_markers,
..Default::default()
},
);
};
insert("bucket", "", 0, 0, 0, 0);
insert("bucket/a", "bucket", 100, 1, 1, 0);
insert("bucket/b", "bucket", 0, 0, 0, 0);
insert("bucket/b/c", "bucket/b", 200, 2, 2, 1);
insert("bucket/b/d", "bucket/b", 40, 1, 3, 0);
insert("bucket/loose", "bucket", 10, 1, 1, 1);
cache
}
#[test]
fn prefix_usage_aggregates_bucket_root_and_one_level_below() {
let cache = prefix_usage_fixture_cache();
let root = cache
.prefix_usage("bucket", "", 100)
.expect("root query must find the bucket entry");
assert_eq!(root.usage.size, 350, "root aggregate flattens the whole subtree");
assert_eq!(root.usage.objects, 5);
assert_eq!(root.usage.versions, 7);
assert_eq!(root.usage.delete_markers, 2);
assert!(!root.compacted);
assert!(!root.truncated);
// Breakdown is one level: b (240) before a (100) before loose (10),
// each flattened to its own subtree total.
let names: Vec<(&str, u64)> = root
.sub_prefixes
.iter()
.map(|entry| (entry.prefix.as_str(), entry.usage.size))
.collect();
assert_eq!(names, vec![("b", 240), ("a", 100), ("loose", 10)]);
}
#[test]
fn prefix_usage_drills_into_arbitrary_prefixes() {
let cache = prefix_usage_fixture_cache();
let b = cache.prefix_usage("bucket", "b", 100).expect("nested prefix must resolve");
assert_eq!(b.usage.size, 240);
assert_eq!(b.usage.versions, 5);
let names: Vec<&str> = b.sub_prefixes.iter().map(|entry| entry.prefix.as_str()).collect();
assert_eq!(names, vec!["c", "d"]);
// Prefix slashes are normalized away.
let slashed = cache.prefix_usage("bucket", "/b/", 100).expect("slash-insensitive lookup");
assert_eq!(slashed.usage.size, 240);
assert!(cache.prefix_usage("bucket", "absent", 100).is_none(), "unknown prefix must be a miss");
assert!(cache.prefix_usage("other", "", 100).is_none(), "unknown bucket must be a miss");
}
#[test]
fn prefix_usage_reports_and_respects_truncation() {
let cache = prefix_usage_fixture_cache();
let capped = cache.prefix_usage("bucket", "", 2).expect("root query");
assert!(capped.truncated, "three children capped to two must flag truncation");
let names: Vec<&str> = capped.sub_prefixes.iter().map(|entry| entry.prefix.as_str()).collect();
assert_eq!(names, vec!["b", "a"], "largest prefixes survive the cut");
}
#[test]
fn prefix_usage_marks_compacted_entries() {
let mut cache = DataUsageCache::default();
cache.replace(
"bucket",
"",
DataUsageEntry {
size: 999,
objects: 9,
compacted: true,
..Default::default()
},
);
let compacted = cache.prefix_usage("bucket", "", 100).expect("compacted root resolves");
assert!(compacted.compacted, "compaction must be visible to callers");
assert_eq!(compacted.usage.size, 999);
assert!(compacted.sub_prefixes.is_empty(), "a compacted entry carries no children");
}
#[test]
fn prefix_usage_rejects_cyclic_and_dangling_caches() {
// A self-referencing child (corrupt cache) must yield a miss for the
// whole query, not unbounded recursion.
let mut cache = prefix_usage_fixture_cache();
if let Some(entry) = cache.cache.get_mut("bucket/b") {
entry.children.insert("bucket/b".to_string());
}
assert!(cache.prefix_usage("bucket", "b", 100).is_none(), "a cyclic subtree must be rejected");
// The unaffected sibling still answers.
assert!(cache.prefix_usage("bucket", "a", 100).is_some());
// A child key with no entry (dangling link) is rejected rather than
// silently dropped: half a tree would under-report usage.
let mut dangling = prefix_usage_fixture_cache();
if let Some(entry) = dangling.cache.get_mut("bucket/b") {
entry.children.insert("bucket/b/ghost".to_string());
}
assert!(
dangling.prefix_usage("bucket", "b", 100).is_none(),
"a dangling child link must be rejected"
);
}
#[test]
fn hash_path_uses_portable_slash_semantics() {
for (input, expected) in [
+1 -4
View File
@@ -67,10 +67,7 @@ fn configured_capture_log_path(temp_dir: &str) -> Option<String> {
capture_log_path(Path::new(&log_dir), temp_dir).map(|path| path.to_string_lossy().into_owned())
}
pub(crate) fn capture_command_logs(
command: &mut Command,
log_path: Option<&str>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
fn capture_command_logs(command: &mut Command, log_path: Option<&str>) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let Some(log_path) = log_path else {
return Ok(());
};
+3 -663
View File
@@ -2,7 +2,6 @@
use crate::common::{RustFSTestEnvironment, init_logging, rustfs_binary_path};
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
use serial_test::serial;
use std::fs;
use std::path::PathBuf;
@@ -26,15 +25,6 @@ fn generate_compressible_data(size: usize) -> Vec<u8> {
data
}
/// Deterministic 2048-byte-period binary pattern that compresses extremely well: every part
/// yields many compressed blocks, which is exactly the shape that reproduced the mid-payload
/// Pending truncation (rustfs/rustfs#5957).
fn generate_high_ratio_binary_data(size: usize, seed: u8) -> Vec<u8> {
(0..size)
.map(|i| ((i as u64).wrapping_mul(2_654_435_761).wrapping_add(seed as u64) >> 3) as u8)
.collect()
}
fn find_part_files(temp_dir: &str, bucket: &str, object_key: &str) -> Vec<PathBuf> {
let bucket_path = PathBuf::from(temp_dir).join(bucket);
let mut part_files = Vec::new();
@@ -65,14 +55,9 @@ async fn start_rustfs_with_compression(env: &mut RustFSTestEnvironment) -> Resul
env.cleanup_existing_processes().await?;
let binary_path = rustfs_binary_path();
// Route the child's stdout/stderr through the shared RUSTFS_E2E_LOG_DIR
// capture (survives the temp-dir cleanup on Drop and is uploaded as a CI
// artifact); without the env var the child inherits stdio as before.
let mut command = Command::new(&binary_path);
command
let process = Command::new(&binary_path)
.env("RUSTFS_CONSOLE_ENABLE", "false")
.env("RUSTFS_COMPRESSION_ENABLED", "true")
.env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true")
.args([
"--address",
&env.address,
@@ -81,9 +66,8 @@ async fn start_rustfs_with_compression(env: &mut RustFSTestEnvironment) -> Resul
"--secret-key",
&env.secret_key,
&env.temp_dir,
]);
crate::common::capture_command_logs(&mut command, env.capture_log_path.as_deref())?;
let process = command.spawn()?;
])
.spawn()?;
env.process = Some(process);
@@ -170,647 +154,3 @@ async fn test_compression_roundtrip() -> Result<(), Box<dyn std::error::Error +
env.stop_server();
Ok(())
}
const MULTIPART_COMPRESSION_BUCKET: &str = "compression-multipart-bucket";
const MPU_PART1_SIZE: usize = 5 * 1024 * 1024;
const MPU_PART2_SIZE: usize = 1024 * 1024;
async fn multipart_upload(
client: &aws_sdk_s3::Client,
bucket: &str,
key: &str,
parts: &[&[u8]],
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let create = client.create_multipart_upload().bucket(bucket).key(key).send().await?;
let upload_id = create.upload_id().ok_or("missing upload id")?.to_string();
let mut completed_parts = Vec::with_capacity(parts.len());
for (i, part) in parts.iter().enumerate() {
let part_number = (i + 1) as i32;
let upload = client
.upload_part()
.bucket(bucket)
.key(key)
.upload_id(&upload_id)
.part_number(part_number)
.body(ByteStream::from(part.to_vec()))
.send()
.await?;
completed_parts.push(
CompletedPart::builder()
.part_number(part_number)
.e_tag(upload.e_tag().unwrap_or_default())
.build(),
);
}
client
.complete_multipart_upload()
.bucket(bucket)
.key(key)
.upload_id(&upload_id)
.multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed_parts)).build())
.send()
.await?;
Ok(())
}
async fn fetch_range(
client: &aws_sdk_s3::Client,
bucket: &str,
key: &str,
range: &str,
) -> Result<Vec<u8>, Box<dyn std::error::Error + Send + Sync>> {
let response = client.get_object().bucket(bucket).key(key).range(range).send().await?;
Ok(response.body.collect().await?.into_bytes().to_vec())
}
/// Multipart disk compression roundtrip: parts are written as independent
/// compressed streams and every GET shape must reassemble the original bytes
/// (rustfs/rustfs#5957: multipart uploads previously bypassed disk compression
/// entirely).
#[tokio::test]
#[serial]
async fn test_compression_multipart_roundtrip() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
info!("Starting multipart compression roundtrip test");
let mut env = RustFSTestEnvironment::new().await?;
start_rustfs_with_compression(&mut env).await?;
let client = env.create_s3_client();
env.create_test_bucket(MULTIPART_COMPRESSION_BUCKET).await?;
let object_key = "multipart-compressible.txt";
let part1 = generate_compressible_data(MPU_PART1_SIZE);
let part2 = generate_compressible_data(MPU_PART2_SIZE);
let mut original_data = part1.clone();
original_data.extend_from_slice(&part2);
let total_size = original_data.len();
multipart_upload(&client, MULTIPART_COMPRESSION_BUCKET, object_key, &[&part1, &part2]).await?;
let head_response = client
.head_object()
.bucket(MULTIPART_COMPRESSION_BUCKET)
.key(object_key)
.send()
.await?;
assert_eq!(
head_response.content_length().unwrap_or(0) as usize,
total_size,
"Content-Length should be the logical object size"
);
let part_files = find_part_files(&env.temp_dir, MULTIPART_COMPRESSION_BUCKET, object_key);
assert!(!part_files.is_empty(), "expected on-disk part files for the multipart object");
let total_physical_size: u64 = part_files.iter().filter_map(|p| fs::metadata(p).ok()).map(|m| m.len()).sum();
assert!(
total_physical_size < (total_size / 2) as u64,
"Physical size {total_physical_size} should be well below original size {total_size} (multipart compression applied)"
);
info!("Multipart physical storage size: {total_physical_size} bytes (compressed from {total_size} bytes)");
// Full GET must reassemble both independently compressed parts.
let get_response = client
.get_object()
.bucket(MULTIPART_COMPRESSION_BUCKET)
.key(object_key)
.send()
.await?;
let downloaded = get_response.body.collect().await?.into_bytes();
assert_eq!(downloaded.len(), total_size);
assert_eq!(&downloaded[..], &original_data[..], "full GET data mismatch");
// Range fully inside part 1.
let range_inside_part1 = fetch_range(&client, MULTIPART_COMPRESSION_BUCKET, object_key, "bytes=1024-999423").await?;
assert_eq!(&range_inside_part1[..], &original_data[1024..999424], "part-1 range mismatch");
// Range crossing the part boundary.
let boundary_start = MPU_PART1_SIZE - 128 * 1024;
let boundary_end = MPU_PART1_SIZE + 128 * 1024 - 1;
let range_crossing = fetch_range(
&client,
MULTIPART_COMPRESSION_BUCKET,
object_key,
&format!("bytes={boundary_start}-{boundary_end}"),
)
.await?;
assert_eq!(
&range_crossing[..],
&original_data[boundary_start..boundary_end + 1],
"boundary-crossing range mismatch"
);
// Range fully inside part 2.
let part2_start = MPU_PART1_SIZE + 4096;
let part2_end = MPU_PART1_SIZE + 256 * 1024 - 1;
let range_inside_part2 = fetch_range(
&client,
MULTIPART_COMPRESSION_BUCKET,
object_key,
&format!("bytes={part2_start}-{part2_end}"),
)
.await?;
assert_eq!(
&range_inside_part2[..],
&original_data[part2_start..part2_end + 1],
"part-2 range mismatch"
);
// Suffix range (last 128 KiB, entirely in part 2).
let suffix_len = 128 * 1024;
let suffix = fetch_range(&client, MULTIPART_COMPRESSION_BUCKET, object_key, &format!("bytes=-{suffix_len}")).await?;
assert_eq!(&suffix[..], &original_data[total_size - suffix_len..], "suffix range mismatch");
// partNumber GETs must return each original part.
for (part_number, expected) in [(1, &part1), (2, &part2)] {
let response = client
.get_object()
.bucket(MULTIPART_COMPRESSION_BUCKET)
.key(object_key)
.part_number(part_number)
.send()
.await?;
let body = response.body.collect().await?.into_bytes();
assert_eq!(&body[..], &expected[..], "partNumber={part_number} GET mismatch");
}
info!("Multipart compression roundtrip test passed");
env.delete_test_bucket(MULTIPART_COMPRESSION_BUCKET).await?;
env.stop_server();
Ok(())
}
const MPU_HIGH_RATIO_BUCKET: &str = "compression-mpu-high-ratio-bucket";
/// High-ratio binary multipart payload: the object key is on the compression allow-list, so the
/// disk-compression path runs and each part is stored as many compressed blocks — the shape that
/// reproduced the mid-payload Pending truncation (rustfs/rustfs#5957). Every GET shape must return
/// the exact original bytes, and the stored size must show the data really was compressed.
#[tokio::test]
#[serial]
async fn test_compression_multipart_high_ratio_binary_roundtrip() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
info!("Starting multipart high-ratio binary compression roundtrip test");
let mut env = RustFSTestEnvironment::new().await?;
start_rustfs_with_compression(&mut env).await?;
let client = env.create_s3_client();
env.create_test_bucket(MPU_HIGH_RATIO_BUCKET).await?;
let object_key = "multipart-high-ratio.txt";
let part1 = generate_high_ratio_binary_data(MPU_PART1_SIZE, 7);
let part2 = generate_high_ratio_binary_data(MPU_PART2_SIZE, 61);
let mut original_data = part1.clone();
original_data.extend_from_slice(&part2);
let total_size = original_data.len();
multipart_upload(&client, MPU_HIGH_RATIO_BUCKET, object_key, &[&part1, &part2]).await?;
let head_response = client
.head_object()
.bucket(MPU_HIGH_RATIO_BUCKET)
.key(object_key)
.send()
.await?;
assert_eq!(
head_response.content_length().unwrap_or(0) as usize,
total_size,
"Content-Length should be the logical object size"
);
// This pattern compresses to roughly 1/50 of its logical size, so a comfortably loose 2x
// margin still proves the parts were stored compressed rather than raw or double-encoded.
let part_files = find_part_files(&env.temp_dir, MPU_HIGH_RATIO_BUCKET, object_key);
assert!(!part_files.is_empty(), "expected on-disk part files for the multipart object");
let total_physical_size: u64 = part_files.iter().filter_map(|p| fs::metadata(p).ok()).map(|m| m.len()).sum();
assert!(
total_physical_size < (total_size as u64) / 2,
"Physical size {total_physical_size} should be far below the logical size {total_size} for high-ratio data"
);
info!("High-ratio multipart physical storage size: {total_physical_size} bytes (logical {total_size} bytes)");
info!("step: full GET");
let get_response = client
.get_object()
.bucket(MPU_HIGH_RATIO_BUCKET)
.key(object_key)
.send()
.await?;
let downloaded = get_response.body.collect().await?.into_bytes();
assert_eq!(downloaded.len(), total_size);
assert_eq!(&downloaded[..], &original_data[..], "full GET data mismatch");
// Range crossing the part boundary.
info!("step: boundary range GET");
let boundary_start = MPU_PART1_SIZE - 128 * 1024;
let boundary_end = MPU_PART1_SIZE + 128 * 1024 - 1;
let range_crossing = fetch_range(
&client,
MPU_HIGH_RATIO_BUCKET,
object_key,
&format!("bytes={boundary_start}-{boundary_end}"),
)
.await?;
assert_eq!(
&range_crossing[..],
&original_data[boundary_start..boundary_end + 1],
"boundary-crossing range mismatch"
);
// partNumber GET for the trailing part.
info!("step: partNumber GET");
let part2_response = client
.get_object()
.bucket(MPU_HIGH_RATIO_BUCKET)
.key(object_key)
.part_number(2)
.send()
.await?;
let part2_body = part2_response.body.collect().await?.into_bytes();
assert_eq!(&part2_body[..], &part2[..], "partNumber=2 GET mismatch");
info!("Multipart high-ratio binary compression roundtrip test passed");
env.delete_test_bucket(MPU_HIGH_RATIO_BUCKET).await?;
env.stop_server();
Ok(())
}
const MPU_COPY_COMPRESSION_BUCKET: &str = "compression-mpu-copy-bucket";
const MPU_COPY_SOURCE_SIZE: usize = 6 * 1024 * 1024;
const MPU_COPY_RANGE_LEN: usize = 5 * 1024 * 1024;
/// UploadPartCopy feeds a part from an already stored (and already compressed) object. The copied
/// range must be decompressed on read and re-compressed into the destination part, so the final
/// object has to match "source prefix + uploaded tail" byte for byte.
#[tokio::test]
#[serial]
async fn test_compression_multipart_upload_part_copy_roundtrip() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
info!("Starting multipart upload-part-copy compression roundtrip test");
let mut env = RustFSTestEnvironment::new().await?;
start_rustfs_with_compression(&mut env).await?;
let client = env.create_s3_client();
env.create_test_bucket(MPU_COPY_COMPRESSION_BUCKET).await?;
// Source object: a plain PUT that goes through the single-stream compression path.
let source_key = "copy-source.txt";
let source_data = generate_compressible_data(MPU_COPY_SOURCE_SIZE);
client
.put_object()
.bucket(MPU_COPY_COMPRESSION_BUCKET)
.key(source_key)
.body(ByteStream::from(source_data.clone()))
.send()
.await?;
// Destination object: part 1 copied from the source, part 2 uploaded directly.
let target_key = "copy-target.txt";
let part2 = generate_compressible_data(MPU_PART2_SIZE);
let mut expected_data = source_data[..MPU_COPY_RANGE_LEN].to_vec();
expected_data.extend_from_slice(&part2);
let total_size = expected_data.len();
let create = client
.create_multipart_upload()
.bucket(MPU_COPY_COMPRESSION_BUCKET)
.key(target_key)
.send()
.await?;
let upload_id = create.upload_id().ok_or("missing upload id")?.to_string();
let copy_part = client
.upload_part_copy()
.bucket(MPU_COPY_COMPRESSION_BUCKET)
.key(target_key)
.upload_id(&upload_id)
.part_number(1)
.copy_source(format!("{MPU_COPY_COMPRESSION_BUCKET}/{source_key}"))
.copy_source_range(format!("bytes=0-{}", MPU_COPY_RANGE_LEN - 1))
.send()
.await?;
let copy_etag = copy_part
.copy_part_result()
.and_then(|r| r.e_tag())
.ok_or("missing copy part etag")?
.to_string();
let uploaded_part = client
.upload_part()
.bucket(MPU_COPY_COMPRESSION_BUCKET)
.key(target_key)
.upload_id(&upload_id)
.part_number(2)
.body(ByteStream::from(part2.clone()))
.send()
.await?;
client
.complete_multipart_upload()
.bucket(MPU_COPY_COMPRESSION_BUCKET)
.key(target_key)
.upload_id(&upload_id)
.multipart_upload(
CompletedMultipartUpload::builder()
.parts(CompletedPart::builder().part_number(1).e_tag(copy_etag).build())
.parts(
CompletedPart::builder()
.part_number(2)
.e_tag(uploaded_part.e_tag().unwrap_or_default())
.build(),
)
.build(),
)
.send()
.await?;
let head_response = client
.head_object()
.bucket(MPU_COPY_COMPRESSION_BUCKET)
.key(target_key)
.send()
.await?;
assert_eq!(
head_response.content_length().unwrap_or(0) as usize,
total_size,
"Content-Length should be the logical object size"
);
let part_files = find_part_files(&env.temp_dir, MPU_COPY_COMPRESSION_BUCKET, target_key);
assert!(!part_files.is_empty(), "expected on-disk part files for the copied object");
let total_physical_size: u64 = part_files.iter().filter_map(|p| fs::metadata(p).ok()).map(|m| m.len()).sum();
assert!(
total_physical_size < (total_size / 2) as u64,
"Physical size {total_physical_size} should be well below original size {total_size} (copied part compression applied)"
);
let get_response = client
.get_object()
.bucket(MPU_COPY_COMPRESSION_BUCKET)
.key(target_key)
.send()
.await?;
let downloaded = get_response.body.collect().await?.into_bytes();
assert_eq!(downloaded.len(), total_size);
assert_eq!(&downloaded[..], &expected_data[..], "copied multipart GET data mismatch");
info!("Multipart upload-part-copy compression roundtrip test passed");
env.delete_test_bucket(MPU_COPY_COMPRESSION_BUCKET).await?;
env.stop_server();
Ok(())
}
const MPU_THREE_PARTS_BUCKET: &str = "compression-mpu-three-parts-bucket";
const MPU_THREE_PARTS_TAIL_SIZE: usize = 512 * 1024;
/// Three-part upload with uneven part sizes: each partNumber GET must map back to exactly one
/// compressed part stream, and a suffix range must resolve inside the trailing part.
#[tokio::test]
#[serial]
async fn test_compression_multipart_three_parts_part_number_gets() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
info!("Starting three-part multipart compression partNumber test");
let mut env = RustFSTestEnvironment::new().await?;
start_rustfs_with_compression(&mut env).await?;
let client = env.create_s3_client();
env.create_test_bucket(MPU_THREE_PARTS_BUCKET).await?;
let object_key = "multipart-three-parts.txt";
let part1 = generate_compressible_data(MPU_PART1_SIZE);
let part2 = generate_compressible_data(MPU_PART1_SIZE);
let part3 = generate_compressible_data(MPU_THREE_PARTS_TAIL_SIZE);
let mut original_data = part1.clone();
original_data.extend_from_slice(&part2);
original_data.extend_from_slice(&part3);
let total_size = original_data.len();
multipart_upload(&client, MPU_THREE_PARTS_BUCKET, object_key, &[&part1, &part2, &part3]).await?;
let head_response = client
.head_object()
.bucket(MPU_THREE_PARTS_BUCKET)
.key(object_key)
.send()
.await?;
assert_eq!(
head_response.content_length().unwrap_or(0) as usize,
total_size,
"Content-Length should be the logical object size"
);
let part_files = find_part_files(&env.temp_dir, MPU_THREE_PARTS_BUCKET, object_key);
assert!(!part_files.is_empty(), "expected on-disk part files for the multipart object");
let total_physical_size: u64 = part_files.iter().filter_map(|p| fs::metadata(p).ok()).map(|m| m.len()).sum();
assert!(
total_physical_size < (total_size / 2) as u64,
"Physical size {total_physical_size} should be well below original size {total_size} (multipart compression applied)"
);
// Every partNumber GET must return exactly the bytes of the corresponding uploaded part.
for (part_number, expected) in [(1, &part1), (2, &part2), (3, &part3)] {
let response = client
.get_object()
.bucket(MPU_THREE_PARTS_BUCKET)
.key(object_key)
.part_number(part_number)
.send()
.await?;
let body = response.body.collect().await?.into_bytes();
assert_eq!(&body[..], &expected[..], "partNumber={part_number} GET mismatch");
}
// Suffix range (last 64 KiB) resolves inside the trailing part.
let suffix_len = 64 * 1024;
let suffix = fetch_range(&client, MPU_THREE_PARTS_BUCKET, object_key, &format!("bytes=-{suffix_len}")).await?;
assert_eq!(&suffix[..], &original_data[total_size - suffix_len..], "suffix range mismatch");
info!("Three-part multipart compression partNumber test passed");
env.delete_test_bucket(MPU_THREE_PARTS_BUCKET).await?;
env.stop_server();
Ok(())
}
const MPU_SSE_COMPRESSION_BUCKET: &str = "compression-mpu-sse-bucket";
async fn start_rustfs_with_compression_and_sse(
env: &mut RustFSTestEnvironment,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
use base64::Engine;
env.cleanup_existing_processes().await?;
let binary_path = rustfs_binary_path();
let master_key = base64::engine::general_purpose::STANDARD.encode([0x42u8; 32]);
// Server output goes to a file inside the per-test temp dir so a failing
// run can be diagnosed from the child's logs.
let server_log = std::fs::File::create(format!("{}/server.log", env.temp_dir))?;
let server_log_err = server_log.try_clone()?;
let process = Command::new(&binary_path)
.env("RUSTFS_CONSOLE_ENABLE", "false")
.env("RUSTFS_COMPRESSION_ENABLED", "true")
.env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true")
.env("RUSTFS_SSE_S3_MASTER_KEY", master_key)
.env("RUST_LOG", "rustfs=info,rustfs_ecstore=info")
.stdout(std::process::Stdio::from(server_log))
.stderr(std::process::Stdio::from(server_log_err))
.args([
"--address",
&env.address,
"--access-key",
&env.access_key,
"--secret-key",
&env.secret_key,
&env.temp_dir,
])
.spawn()?;
env.process = Some(process);
info!("Waiting for RustFS server with compression + SSE-S3 enabled on {}", env.address);
for i in 0..30 {
if TcpStream::connect(&env.address).await.is_ok() {
info!("RustFS server is ready after {} attempts", i + 1);
return Ok(());
}
if i == 29 {
return Err("RustFS server failed to become ready".into());
}
sleep(Duration::from_secs(1)).await;
}
Ok(())
}
/// SSE-S3 + disk compression multipart: each part is compressed and then encrypted, and every GET
/// shape must still return the original plaintext bytes. Physical size must shrink because the
/// compression runs before encryption.
#[tokio::test]
#[serial]
async fn test_compression_multipart_sse_s3_roundtrip() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
use aws_sdk_s3::types::ServerSideEncryption;
init_logging();
info!("Starting SSE-S3 multipart compression roundtrip test");
let mut env = RustFSTestEnvironment::new().await?;
start_rustfs_with_compression_and_sse(&mut env).await?;
let client = env.create_s3_client();
env.create_test_bucket(MPU_SSE_COMPRESSION_BUCKET).await?;
let object_key = "multipart-sse-compressible.txt";
let part1 = generate_compressible_data(MPU_PART1_SIZE);
let part2 = generate_compressible_data(MPU_PART2_SIZE);
let mut original_data = part1.clone();
original_data.extend_from_slice(&part2);
let total_size = original_data.len();
let create = client
.create_multipart_upload()
.bucket(MPU_SSE_COMPRESSION_BUCKET)
.key(object_key)
.server_side_encryption(ServerSideEncryption::Aes256)
.send()
.await?;
let upload_id = create.upload_id().ok_or("missing upload id")?.to_string();
let mut completed_parts = Vec::new();
for (i, part) in [&part1, &part2].into_iter().enumerate() {
let part_number = (i + 1) as i32;
let upload = client
.upload_part()
.bucket(MPU_SSE_COMPRESSION_BUCKET)
.key(object_key)
.upload_id(&upload_id)
.part_number(part_number)
.body(ByteStream::from(part.clone()))
.send()
.await?;
completed_parts.push(
CompletedPart::builder()
.part_number(part_number)
.e_tag(upload.e_tag().unwrap_or_default())
.build(),
);
}
client
.complete_multipart_upload()
.bucket(MPU_SSE_COMPRESSION_BUCKET)
.key(object_key)
.upload_id(&upload_id)
.multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed_parts)).build())
.send()
.await?;
let head_response = client
.head_object()
.bucket(MPU_SSE_COMPRESSION_BUCKET)
.key(object_key)
.send()
.await?;
assert_eq!(
head_response.content_length().unwrap_or(0) as usize,
total_size,
"Content-Length should be the logical object size"
);
assert_eq!(
head_response.server_side_encryption(),
Some(&ServerSideEncryption::Aes256),
"HEAD must report SSE-S3"
);
let part_files = find_part_files(&env.temp_dir, MPU_SSE_COMPRESSION_BUCKET, object_key);
assert!(!part_files.is_empty(), "expected on-disk part files for the multipart object");
let total_physical_size: u64 = part_files.iter().filter_map(|p| fs::metadata(p).ok()).map(|m| m.len()).sum();
assert!(
total_physical_size < (total_size / 2) as u64,
"Physical size {total_physical_size} should be well below original size {total_size} (compress-then-encrypt applied)"
);
let get_response = client
.get_object()
.bucket(MPU_SSE_COMPRESSION_BUCKET)
.key(object_key)
.send()
.await?;
let downloaded = get_response.body.collect().await?.into_bytes();
assert_eq!(downloaded.len(), total_size);
assert_eq!(&downloaded[..], &original_data[..], "SSE-S3 multipart full GET data mismatch");
// Range crossing the part boundary must decrypt and decompress across parts.
let boundary_start = MPU_PART1_SIZE - 64 * 1024;
let boundary_end = MPU_PART1_SIZE + 64 * 1024 - 1;
let range_crossing = fetch_range(
&client,
MPU_SSE_COMPRESSION_BUCKET,
object_key,
&format!("bytes={boundary_start}-{boundary_end}"),
)
.await?;
assert_eq!(
&range_crossing[..],
&original_data[boundary_start..boundary_end + 1],
"SSE-S3 boundary-crossing range mismatch"
);
// partNumber GET for the trailing part.
let part2_response = client
.get_object()
.bucket(MPU_SSE_COMPRESSION_BUCKET)
.key(object_key)
.part_number(2)
.send()
.await?;
let part2_body = part2_response.body.collect().await?.into_bytes();
assert_eq!(&part2_body[..], &part2[..], "SSE-S3 partNumber=2 GET mismatch");
info!("SSE-S3 multipart compression roundtrip test passed");
env.delete_test_bucket(MPU_SSE_COMPRESSION_BUCKET).await?;
env.stop_server();
Ok(())
}
+1 -91
View File
@@ -76,18 +76,6 @@ const SOURCE_MTIME_HEADERS: [&str; 2] = ["x-rustfs-source-mtime", "x-minio-sourc
const SOURCE_REPLICATION_REQUEST_HEADERS: [&str; 2] =
["x-rustfs-source-replication-request", "x-minio-source-replication-request"];
const SOURCE_ETAG_HEADERS: [&str; 2] = ["x-rustfs-source-etag", "x-minio-source-etag"];
const SOURCE_TAGGING_TIMESTAMP_HEADERS: [&str; 2] = [
"x-rustfs-source-replication-tagging-timestamp",
"x-minio-source-replication-tagging-timestamp",
];
const SOURCE_RETENTION_TIMESTAMP_HEADERS: [&str; 2] = [
"x-rustfs-source-replication-retention-timestamp",
"x-minio-source-replication-retention-timestamp",
];
const SOURCE_LEGALHOLD_TIMESTAMP_HEADERS: [&str; 2] = [
"x-rustfs-source-replication-legalhold-timestamp",
"x-minio-source-replication-legalhold-timestamp",
];
const RESERVED_BUCKET_PREFIXES: [&str; 3] = ["xn--", "sthree-", "amzn-s3-demo-"];
const RESERVED_BUCKET_SUFFIXES: [&str; 6] = ["-s3alias", "--ol-s3", ".mrap", "--x-s3", "--table-s3", "-an"];
@@ -130,25 +118,6 @@ pub enum FaultAction {
WrongEtag,
}
/// Replication LWW timestamp headers observed on a request, journaled so
/// sender-side tests can assert what a real target would receive.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ReplicationTimestampHeaders {
pub tagging: Option<String>,
pub retention: Option<String>,
pub legalhold: Option<String>,
}
impl ReplicationTimestampHeaders {
fn from_headers(headers: &HeaderMap) -> Self {
Self {
tagging: header_value(headers, &SOURCE_TAGGING_TIMESTAMP_HEADERS).map(bounded_journal_value),
retention: header_value(headers, &SOURCE_RETENTION_TIMESTAMP_HEADERS).map(bounded_journal_value),
legalhold: header_value(headers, &SOURCE_LEGALHOLD_TIMESTAMP_HEADERS).map(bounded_journal_value),
}
}
}
/// Credential-free request metadata retained for deterministic assertions.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RequestRecord {
@@ -162,7 +131,6 @@ pub struct RequestRecord {
pub part_number: Option<i32>,
pub content_length: Option<u64>,
pub consumed_bytes: Option<usize>,
pub replication_timestamps: ReplicationTimestampHeaders,
pub fault: Option<FaultAction>,
}
@@ -568,15 +536,7 @@ impl S3Access for FaultAccess {
.get(CONTENT_LENGTH)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.parse().ok());
let replication_timestamps = ReplicationTimestampHeaders::from_headers(context.headers());
let fault = record_request(
&self.control,
operation,
context.method().clone(),
parsed,
content_length,
replication_timestamps,
);
let fault = record_request(&self.control, operation, context.method().clone(), parsed, content_length);
if let Some(RequestFault {
action: FaultAction::Status(status),
..
@@ -629,7 +589,6 @@ fn record_request(
method: Method,
parsed: ParsedRequest,
content_length: Option<u64>,
replication_timestamps: ReplicationTimestampHeaders,
) -> Option<RequestFault> {
let mut state = lock(control);
let action = parsed
@@ -654,7 +613,6 @@ fn record_request(
part_number: parsed.part_number,
content_length,
consumed_bytes: None,
replication_timestamps,
fault: action.clone(),
});
action.map(|action| RequestFault { sequence, action })
@@ -1741,52 +1699,6 @@ mod tests {
.await?)
}
#[tokio::test]
async fn journals_replication_timestamp_headers() -> Result<(), BoxError> {
let target = FakeS3Target::start().await?;
target.create_bucket("target-bucket");
let client = client(&target);
client
.put_object()
.bucket("target-bucket")
.key("plain")
.body(ByteStream::from_static(b"plain"))
.send()
.await?;
client
.put_object()
.bucket("target-bucket")
.key("stamped")
.body(ByteStream::from_static(b"stamped"))
.customize()
.map_request(move |mut request| {
let headers = request.headers_mut();
headers.insert("x-rustfs-source-replication-tagging-timestamp", "2026-01-02T03:04:05Z");
headers.insert("x-minio-source-replication-retention-timestamp", "2026-01-02T03:04:06Z");
headers.insert("x-rustfs-source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z");
Ok::<_, std::convert::Infallible>(request)
})
.send()
.await?;
let requests = target.requests();
let plain = requests
.iter()
.find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("plain"))
.expect("plain PUT must be journaled");
assert_eq!(plain.replication_timestamps, ReplicationTimestampHeaders::default());
let stamped = requests
.iter()
.find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("stamped"))
.expect("stamped PUT must be journaled");
assert_eq!(stamped.replication_timestamps.tagging.as_deref(), Some("2026-01-02T03:04:05Z"));
assert_eq!(stamped.replication_timestamps.retention.as_deref(), Some("2026-01-02T03:04:06Z"));
assert_eq!(stamped.replication_timestamps.legalhold.as_deref(), Some("2026-01-02T03:04:07Z"));
Ok(())
}
macro_rules! assert_sdk_error {
($error:expr, $status:expr, $code:expr) => {{
let error = &$error;
@@ -3073,7 +2985,6 @@ mod tests {
part_number: None,
},
Some(0),
ReplicationTimestampHeaders::default(),
);
}
let records = lock(&control).requests.clone();
@@ -3095,7 +3006,6 @@ mod tests {
part_number: None,
},
None,
ReplicationTimestampHeaders::default(),
);
{
let bounded_records = lock(&bounded_control);
@@ -1828,36 +1828,33 @@ async fn four_node_compressed_inline_fallback() -> TestResult {
Ok(())
}
/// Multipart disk compression is live again, so a compression-enabled cluster classifies multipart objects as compressed and the roundtrip (full GET plus partNumber GET) must still return the original bytes.
/// Reverting the multipart compression fix must fail this test.
#[tokio::test]
#[serial]
async fn four_node_multipart_disk_compression_roundtrip() -> TestResult {
async fn four_node_multipart_ignores_disk_compression_fallback() -> TestResult {
init_logging();
let collector = OtlpMetricCollector::start().await?;
let mut cluster = RustFSTestClusterEnvironment::new(4).await?;
configure_reader_metric_cluster(&mut cluster, &collector);
cluster.set_env("RUSTFS_COMPRESSION_ENABLED", "true");
cluster.set_env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true");
cluster.start().await?;
let bucket = "inline-multipart-compression-roundtrip";
let bucket = "inline-multipart-compression-fallback";
cluster.create_test_bucket(bucket).await?;
let client = cluster.create_s3_client(0)?;
let key = "multipart/compressed.txt";
let key = "multipart/compression-disabled.txt";
let (body, second_part, etag) = put_two_part_multipart(&client, bucket, key).await?;
assert_reader_path(
&collector,
&client,
ReaderPathExpectation::for_class(ReaderObject::new(bucket, key, &body, etag.as_deref(), None), LEGACY_DUPLEX, COMPRESSED),
ReaderPathExpectation::for_class(ReaderObject::new(bucket, key, &body, etag.as_deref(), None), LEGACY_DUPLEX, MULTIPART),
)
.await?;
assert_part_number_reader_path(
&collector,
&client,
PartNumberReaderPathExpectation::new(bucket, key, &second_part, body.len(), COMPRESSED, LEGACY_DUPLEX),
PartNumberReaderPathExpectation::new(bucket, key, &second_part, body.len(), MULTIPART, LEGACY_DUPLEX),
)
.await?;
@@ -1874,7 +1871,6 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> Te
let sse_master_key = base64::engine::general_purpose::STANDARD.encode([0x42u8; 32]);
cluster.set_env("RUSTFS_SSE_S3_MASTER_KEY", sse_master_key);
cluster.set_env("RUSTFS_COMPRESSION_ENABLED", "true");
cluster.set_env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true");
configure_mixed_msgpack_cluster(&mut cluster, &collector)?;
cluster.start().await?;
@@ -1894,21 +1890,14 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> Te
ReaderPathExpectation::for_class(
ReaderObject::new(bucket, multipart_key, &multipart_body, multipart_etag.as_deref(), None),
LEGACY_DUPLEX,
COMPRESSED,
MULTIPART,
),
)
.await?;
assert_part_number_reader_path(
&collector,
&client,
PartNumberReaderPathExpectation::new(
bucket,
multipart_key,
&second_part,
multipart_body.len(),
COMPRESSED,
LEGACY_DUPLEX,
),
PartNumberReaderPathExpectation::new(bucket, multipart_key, &second_part, multipart_body.len(), MULTIPART, LEGACY_DUPLEX),
)
.await?;
assert_msgpack_decode_observed(&collector, &decode_before).await?;
@@ -2364,11 +2353,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_
hot_client.create_bucket().bucket(bucket).send().await?;
put_lifecycle_with_transition_retry(&hot_client, bucket, &tier_name).await?;
// `.zip` sits on the disk-compression exclusion list: this test pins
// msgpack compat controls across ILM transition, and a compressed object
// would classify as `compressed` instead of `remote` (and the warm-tier
// read path does not decode compression — tracked separately).
let key = "transition/mixed-multipart.zip";
let key = "transition/mixed-multipart.bin";
let (body, second_part, etag) = put_two_part_multipart(&hot_client, bucket, key).await?;
wait_for_transition(&hot_client, bucket, key, &tier_name).await?;
assert!(
@@ -97,7 +97,7 @@ async fn start_enforcing_ilm_server(env: &mut LocalKMSTestEnvironment) -> TestRe
let envs = [
("RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS", "true"),
("RUSTFS_KMS_ENFORCE_SSE_KEY_POLICY", "true"),
("RUSTFS_KMS_ENFORCE_SSE_KEY_POLICY", "false"),
("RUSTFS_SCANNER_CYCLE", "1"),
("RUSTFS_ILM_PROCESS_TIME", "1"),
("RUSTFS_ILM_DEBUG_DAY_SECS", "2"),
@@ -486,6 +486,7 @@ async fn ilm_expiration_on_sse_kms_bucket_under_enforcement() -> TestResult {
/// depend on scanner scheduling; the 1s scanner cycle stays on as a backstop.
#[tokio::test]
#[serial]
#[ignore = "pins rustfs/rustfs#6025: GET on a transitioned managed-SSE object silently returns corrupt bytes (fails with enforcement on AND off, so it is not an authorization regression); un-ignore with the fix"]
async fn ilm_transition_on_sse_kms_bucket_under_enforcement_reads_back() -> TestResult {
init_logging();
+1
View File
@@ -273,6 +273,7 @@ proptest = "1"
rcgen.workspace = true
insta = { workspace = true, features = ["yaml", "json"] }
rustfs-crypto = { workspace = true }
tonic-prost = { workspace = true }
[build-dependencies]
shadow-rs = { workspace = true, default-features = false, features = ["build", "metadata"] }
+15 -19
View File
@@ -135,8 +135,7 @@ pub mod bucket {
pub use crate::bucket::metadata_sys::ConfigWriteLockProbe;
pub use crate::bucket::metadata_sys::{
BucketMetadataMutationGuard, BucketMetadataSys, ObjectLockConfigState, acquire_bucket_metadata_transaction_lock,
acquire_bucket_metadata_transaction_lock_for_incarnation, capture_bucket_metadata_incarnation, delete,
delete_if_incarnation, delete_under_transaction_lock, get, get_accelerate_config, get_bucket_policy,
capture_bucket_metadata_incarnation, delete, delete_if_incarnation, get, get_accelerate_config, get_bucket_policy,
get_bucket_policy_raw, get_bucket_targets_config, get_config_from_disk, get_cors_config, get_durability_config,
get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config, get_notification_config,
get_object_lock_config, get_object_lock_config_state, get_public_access_block_config, get_quota_config,
@@ -185,18 +184,17 @@ pub mod bucket {
mrf_backlog_observability_snapshot,
};
pub use crate::bucket::replication::{
BucketReplicationResyncStatus, BucketReplicationStat, BucketReplicationStats, BucketStats,
DeleteReplicationConfigSnapshot, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, InQueueMetric,
MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION,
REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATE_INCOMING_DELETE,
REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS,
ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig,
ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationDeleteScheduleInput,
ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO,
ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge,
ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError,
ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus,
VersionPurgeStatusType, XferStats, commit_force_delete_intent, complete_force_delete_intent,
BucketReplicationResyncStatus, BucketReplicationStats, BucketStats, DeleteReplicationConfigSnapshot,
DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, MrfOpKind, MrfReplicateEntry,
MustReplicateOptions, ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS,
REMOTE_TARGET_WRITABLE_FIELDS, REPLICATE_INCOMING_DELETE, REPLICATION_CAPABILITY_CONTRACT_VERSION,
REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicateDecision, ReplicateObjectInfo,
ReplicationBatchAdmission, ReplicationConfig, ReplicationConfigStructureError, ReplicationConfigurationExt,
ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge,
ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission,
ReplicationScannerBridge, ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage,
ReplicationTargetValidationError, ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog,
TargetReplicationResyncStatus, VersionPurgeStatusType, commit_force_delete_intent, complete_force_delete_intent,
delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool,
get_global_replication_stats, init_background_replication, invalid_replication_config_status_field,
persist_force_delete_intent, read_durable_mrf_backlog, replication_state_to_filemeta, replication_status_to_filemeta,
@@ -280,9 +278,7 @@ pub mod cluster {
}
pub mod compression {
pub use crate::io_support::compress::{
MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible, is_disk_compression_enabled, is_multipart_disk_compression_enabled,
};
pub use crate::io_support::compress::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible, is_disk_compression_enabled};
}
pub mod config {
@@ -373,8 +369,8 @@ pub mod error {
pub mod erasure {
pub use crate::erasure::coding::{
BitrotReader, BitrotSelfTestError, BitrotWriter, BitrotWriterWrapper, CustomWriter, Erasure, ErasureConstructionError,
ReedSolomonEncoder, bitrot_self_test, calc_shard_size, calc_shard_size_legacy,
BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, Erasure, ErasureConstructionError, ReedSolomonEncoder,
calc_shard_size, calc_shard_size_legacy,
};
}
+6 -73
View File
@@ -58,9 +58,7 @@ use rustfs_utils::http::{
};
use rustfs_utils::http::{
SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK,
SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST,
SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID,
insert_header,
SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, insert_header,
};
use rustls_pki_types::pem::PemObject;
use serde::{Deserialize, Serialize};
@@ -82,6 +80,7 @@ use tracing::warn;
use url::Url;
use uuid::Uuid;
const DEFAULT_HEALTH_CHECK_RELOAD_DURATION: Duration = Duration::from_secs(30 * 60);
const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16;
const REDACTED_CREDENTIAL: &str = "<redacted>";
@@ -1477,12 +1476,9 @@ impl Default for AdvancedPutOptions {
replication_status: ReplicationStatusType::Pending,
source_mtime: OffsetDateTime::now_utc(),
replication_request: false,
// UNIX_EPOCH means "never modified": header() must not emit a
// timestamp header for it, otherwise a receiver would treat an
// unset category as a modification made right now.
retention_timestamp: OffsetDateTime::UNIX_EPOCH,
tagging_timestamp: OffsetDateTime::UNIX_EPOCH,
legalhold_timestamp: OffsetDateTime::UNIX_EPOCH,
retention_timestamp: OffsetDateTime::now_utc(),
tagging_timestamp: OffsetDateTime::now_utc(),
legalhold_timestamp: OffsetDateTime::now_utc(),
replication_validity_check: false,
}
}
@@ -1549,8 +1545,8 @@ impl Default for PutObjectOptions {
}
}
#[allow(dead_code)]
impl PutObjectOptions {
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
fn set_match_etag(&mut self, etag: &str) {
if etag == "*" {
self.custom_header
@@ -1561,7 +1557,6 @@ impl PutObjectOptions {
}
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
fn set_match_etag_except(&mut self, etag: &str) {
if etag == "*" {
self.custom_header
@@ -1680,16 +1675,6 @@ impl PutObjectOptions {
);
}
for (suffix, timestamp) in [
(SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, self.internal.tagging_timestamp),
(SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, self.internal.retention_timestamp),
(SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, self.internal.legalhold_timestamp),
] {
if timestamp.unix_timestamp() != 0 {
insert_header(&mut header, suffix, timestamp.format(&Rfc3339).unwrap_or_default());
}
}
if self.internal.replication_request {
insert_header(&mut header, SUFFIX_SOURCE_REPLICATION_REQUEST, "true");
}
@@ -1697,7 +1682,6 @@ impl PutObjectOptions {
header
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
fn validate(&self, _c: Arc<TargetClient>) -> Result<(), std::io::Error> {
//if self.checksum.is_set() {
/*if !self.trailing_header_support {
@@ -2858,57 +2842,6 @@ mod tests {
);
}
#[test]
fn put_object_headers_carry_replication_timestamp_headers() {
// MinIO receivers resolve concurrent tag/retention/legal-hold edits by
// last-writer-wins on these headers (object-api-options.go parses them
// as RFC3339); a replica without them loses every conflict resolution.
let mut opts = PutObjectOptions::default();
opts.internal.replication_request = true;
let tagging = OffsetDateTime::from_unix_timestamp(1_700_000_001).expect("valid timestamp");
let retention = OffsetDateTime::from_unix_timestamp(1_700_000_002).expect("valid timestamp");
let legalhold = OffsetDateTime::from_unix_timestamp(1_700_000_003).expect("valid timestamp");
opts.internal.tagging_timestamp = tagging;
opts.internal.retention_timestamp = retention;
opts.internal.legalhold_timestamp = legalhold;
let header = opts.header();
for (suffix, expected) in [
("source-replication-tagging-timestamp", tagging),
("source-replication-retention-timestamp", retention),
("source-replication-legalhold-timestamp", legalhold),
] {
assert_eq!(
rustfs_utils::http::get_header(&header, suffix).as_deref(),
Some(expected.format(&Rfc3339).expect("RFC3339 timestamp").as_str()),
"replication put requests must carry the {suffix} header"
);
}
}
#[test]
fn put_object_headers_omit_unset_replication_timestamps() {
// UNIX_EPOCH means "never modified on the source"; sending it would
// make the receiver treat an unset category as a fresh modification.
let mut opts = PutObjectOptions::default();
opts.internal.replication_request = true;
opts.internal.tagging_timestamp = OffsetDateTime::UNIX_EPOCH;
opts.internal.retention_timestamp = OffsetDateTime::UNIX_EPOCH;
opts.internal.legalhold_timestamp = OffsetDateTime::UNIX_EPOCH;
let header = opts.header();
for suffix in [
"source-replication-tagging-timestamp",
"source-replication-retention-timestamp",
"source-replication-legalhold-timestamp",
] {
assert!(
rustfs_utils::http::get_header(&header, suffix).is_none(),
"unset {suffix} must not be sent to replication targets"
);
}
}
#[tokio::test]
async fn get_remote_target_client_internal_rejects_loopback_endpoint() {
let sys = BucketTargetSys::default();
@@ -46,13 +46,15 @@ use crate::bucket::lifecycle::transition_transaction::run_transition_transaction
use crate::bucket::object_lock::ObjectLockApi;
use crate::bucket::versioning::VersioningApi as _;
use crate::bucket::versioning_sys::BucketVersioningSys;
use crate::client::object_api_utils::new_getobjectreader;
use crate::disk::error::DiskError;
use crate::disk::{DeleteOptions, Disk, DiskAPI, RUSTFS_META_BUCKET, RUSTFS_META_MULTIPART_BUCKET, STORAGE_FORMAT_FILE};
use crate::error::Error;
use crate::error::StorageError;
use crate::error::{is_err_object_not_found, is_err_read_quorum, is_err_version_not_found, is_network_or_host_down};
use crate::error::{
error_resp_to_object_err, is_err_object_not_found, is_err_read_quorum, is_err_version_not_found, is_network_or_host_down,
};
use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions};
use crate::object_api::{ObjectEncryptionResolver, ReadPlan};
use crate::services::tier::{
tier::{TierConfigMgr, TierOperationLease, tier_destination_id_from_metadata},
warm_backend::WarmBackendGetOpts,
@@ -126,23 +128,11 @@ const EVENT_LIFECYCLE_EXPIRED_DETECTED: &str = "lifecycle_expired_detected";
const EVENT_LIFECYCLE_NOT_ENQUEUED: &str = "lifecycle_not_enqueued";
const EVENT_LIFECYCLE_DELETE_DISPATCHED: &str = "lifecycle_delete_dispatched";
const EVENT_LIFECYCLE_DELETE_COMPLETED: &str = "lifecycle_delete_completed";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
const EVENT_LIFECYCLE_TIER_AUDIT: &str = "lifecycle_tier_audit";
const EVENT_LIFECYCLE_TIER_OPERATION_FAILED: &str = "lifecycle_tier_operation_failed";
const EVENT_LIFECYCLE_DELETE_FAILED: &str = "lifecycle_delete_failed";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub type TimeFn = Arc<dyn Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub type TraceFn =
Arc<dyn Fn(String, HashMap<String, String>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
pub type ExpiryOpType = Box<dyn ExpiryOp + Send + Sync + 'static>;
@@ -152,21 +142,9 @@ static TIER_FREE_VERSION_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
static MANUAL_TRANSITION_JOB_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
pub const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub const AMZ_TAG_COUNT: &str = "x-amz-tagging-count";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub const AMZ_TAG_DIRECTIVE: &str = "X-Amz-Tagging-Directive";
pub const AMZ_ENCRYPTION_AES: &str = "AES256";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub const AMZ_ENCRYPTION_KMS: &str = "aws:kms";
pub const ERR_INVALID_STORAGECLASS: &str = "invalid tier.";
@@ -304,10 +282,6 @@ impl LifecycleSys {
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub fn trace(oi: &ObjectInfo) -> TraceFn {
let bucket = oi.bucket.clone();
let name = oi.name.clone();
@@ -456,23 +430,16 @@ impl<'a> LifecycleExpiryTrace<'a> {
}
}
#[allow(dead_code)]
impl ExpiryStats {
pub fn missed_tasks(&self) -> i64 {
self.missed_expiry_tasks.load(Ordering::SeqCst)
}
#[allow(
dead_code,
reason = "asserted by this file's tests; the lib target cannot see test-only consumers (backlog#1823)"
)]
fn missed_free_vers_tasks(&self) -> i64 {
self.missed_freevers_tasks.load(Ordering::SeqCst)
}
#[allow(
dead_code,
reason = "asserted by this file's tests; the lib target cannot see test-only consumers (backlog#1823)"
)]
fn missed_tier_journal_tasks(&self) -> i64 {
self.missed_tier_journal_tasks.load(Ordering::SeqCst)
}
@@ -605,10 +572,6 @@ async fn delete_free_version_remote_object(
Ok(())
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
async fn delete_free_version_remote_object_then<T, F, Fut>(
oi: &ObjectInfo,
tier_config_mgr: &Arc<RwLock<TierConfigMgr>>,
@@ -2907,10 +2870,6 @@ fn stale_upload_default_due(initiated: OffsetDateTime, default_expiry: StdDurati
initiated + time::Duration::seconds(default_expiry.as_secs() as i64)
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
async fn stale_upload_current_size(set: &Arc<SetDisks>, metadata: &HashMap<String, String>, upload_dir: &str) -> Option<usize> {
stale_upload_current_size_with_opts(set, metadata, upload_dir, false).await
}
@@ -3395,10 +3354,6 @@ pub async fn validate_transition_tier(lc: &BucketLifecycleConfiguration) -> Resu
Ok(())
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
fn mark_delete_opts_skip_decommissioned_on_remote_success(opts: &mut ObjectOptions, remote_delete_succeeded: bool) {
if remote_delete_succeeded {
opts.skip_decommissioned = true;
@@ -4386,10 +4341,6 @@ pub async fn expire_transitioned_object(
Ok(dobj)
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub fn gen_transition_objname(bucket: &str) -> Result<String, Error> {
let us = Uuid::new_v4().to_string();
let mut hasher = Sha256::new();
@@ -4424,10 +4375,6 @@ pub async fn transition_object(api: Arc<ECStore>, oi: &ObjectInfo, lae: LcAuditE
result
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub fn audit_tier_actions(_tier: &str, bytes: i64) -> TimeFn {
let tier = _tier.to_string();
Arc::new(move || {
@@ -4446,10 +4393,6 @@ pub fn audit_tier_actions(_tier: &str, bytes: i64) -> TimeFn {
})
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn get_transitioned_object_reader(
bucket: &str,
object: &str,
@@ -4457,10 +4400,9 @@ pub async fn get_transitioned_object_reader(
h: &HeaderMap,
oi: &ObjectInfo,
opts: &ObjectOptions,
resolver: Option<&dyn ObjectEncryptionResolver>,
) -> Result<GetObjectReader, std::io::Error> {
let tier_config_mgr = runtime_sources::tier_config_mgr_handle();
get_transitioned_object_reader_with_tier_manager(bucket, object, rs, h, oi, opts, &tier_config_mgr, resolver).await
get_transitioned_object_reader_with_tier_manager(bucket, object, rs, h, oi, opts, &tier_config_mgr).await
}
fn validate_transition_remote_version(oi: &ObjectInfo) -> Result<bool, std::io::Error> {
@@ -4480,10 +4422,6 @@ fn validate_transition_remote_version(oi: &ObjectInfo) -> Result<bool, std::io::
}
}
// The resolver joins the tier manager as the second injected port this read
// needs; grouping the request half into a struct would churn every call site of
// a bug fix.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn get_transitioned_object_reader_with_tier_manager(
bucket: &str,
object: &str,
@@ -4492,7 +4430,6 @@ pub(crate) async fn get_transitioned_object_reader_with_tier_manager(
oi: &ObjectInfo,
opts: &ObjectOptions,
tier_config_mgr: &Arc<RwLock<TierConfigMgr>>,
resolver: Option<&dyn ObjectEncryptionResolver>,
) -> Result<GetObjectReader, std::io::Error> {
validate_transition_remote_version(oi)?;
let expected_identity = tier_destination_id_from_metadata(&oi.user_defined)?;
@@ -4510,16 +4447,11 @@ pub(crate) async fn get_transitioned_object_reader_with_tier_manager(
tgt_client.validate_remote_version_id(&oi.transitioned_object.version_id)?;
// The same read plan the local path uses, so the tier fetch is positioned in
// the object's *stored* coordinate system and the stream is handed the same
// decrypt/decompress transforms. Reading an encrypted object's ciphertext
// through a plaintext-coordinate range and skipping the transform is how a
// transitioned SSE object used to come back as silently corrupt bytes of the
// right length (rustfs/rustfs#6025).
let plan = ReadPlan::build_for_request(rs.clone(), oi, opts, h, resolver)
.await
.map_err(|err| std::io::Error::other(format!("building the read plan for {bucket}/{object} failed: {err}")))?;
let (off, length) = (plan.storage_offset() as i64, plan.storage_length());
let ret = new_getobjectreader(rs, oi, opts, h);
if let Err(err) = ret {
return Err(error_resp_to_object_err(err, vec![bucket, object]));
}
let (get_fn, off, length) = ret.expect("get_transitioned_object_reader should succeed after error check");
let mut gopts = WarmBackendGetOpts::default();
if off >= 0 && length >= 0 {
@@ -4556,10 +4488,7 @@ pub(crate) async fn get_transitioned_object_reader_with_tier_manager(
);
e
})?;
let object_reader = plan
.into_object_reader(Box::new(reader), oi)
.map_err(|err| std::io::Error::other(format!("wrapping the tier stream for {bucket}/{object} failed: {err}")))?;
Ok(attach_tier_operation_lease(object_reader, tgt_client))
Ok(attach_tier_operation_lease(get_fn(reader, h.clone()), tgt_client))
}
struct TierOperationLeaseReader {
@@ -5204,10 +5133,6 @@ async fn lifecycle_delete_config_snapshot(api: &ECStore, oi: &ObjectInfo) -> Res
ReplicationObjectBridge::delete_request_config(api, &oi.bucket).await
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
let mut success = false;
match event.action {
@@ -5851,7 +5776,6 @@ mod tests {
&object_info,
&ObjectOptions::default(),
&manager,
None,
)
.await
.expect("transitioned reader should open");
@@ -5916,7 +5840,6 @@ mod tests {
&object_info,
&ObjectOptions::default(),
&manager,
None,
)
.await
{
@@ -5957,7 +5880,6 @@ mod tests {
&object_info,
&ObjectOptions::default(),
&manager,
None,
)
.await
{
@@ -6195,7 +6117,6 @@ mod tests {
&oi,
&ObjectOptions::default(),
&manager,
None,
)
.await
{
@@ -6219,7 +6140,6 @@ mod tests {
&oi,
&ObjectOptions::default(),
&manager,
None,
)
.await
{
@@ -7485,10 +7405,6 @@ mod tests {
// process environment while `env::set_var`/`env::remove_var` is active.
// SAFETY: keep this note adjacent to the allowance for the repository guard.
#[allow(unsafe_code)]
#[allow(
dead_code,
reason = "transition-queue env fixture kept for tests that scope those vars; no test uses it today (backlog#1823)"
)]
async fn with_transition_queue_env_async<F, Fut>(capacity: Option<&str>, timeout_ms: Option<&str>, test_fn: F)
where
F: FnOnce() -> Fut,
@@ -759,10 +759,6 @@ pub struct ManualTransitionWorkerResultRecord {
}
impl ManualTransitionWorkerResultRecord {
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub fn new(job_id: Uuid, task_key: impl Into<String>, result: ManualTransitionWorkerResult) -> Self {
Self::new_with_reason(job_id, task_key, result, None)
}
@@ -1261,10 +1257,6 @@ pub(crate) async fn save_manual_transition_task_if_absent(
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn load_manual_transition_task_record(
api: Arc<ECStore>,
job_id: Uuid,
@@ -1328,10 +1320,6 @@ async fn scan_manual_transition_task_journal(api: Arc<ECStore>, job_id: Uuid) ->
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn load_manual_transition_worker_result_stats(
api: Arc<ECStore>,
job_id: Uuid,
@@ -1467,10 +1455,6 @@ async fn scan_manual_transition_worker_result_journal(
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn reconcile_manual_transition_worker_results(
api: Arc<ECStore>,
job_id: Uuid,
@@ -15,35 +15,25 @@
use rustfs_common::metrics::IlmAction;
use crate::bucket::lifecycle::lifecycle::ObjectOpts;
use crate::bucket::replication::ReplicationLifecycleBridge;
pub(crate) use crate::bucket::replication::ReplicationStatusType;
#[cfg(test)]
pub(crate) use crate::bucket::replication::VersionPurgeStatusType;
pub(crate) use crate::bucket::replication::{
DeleteReplicationConfigSnapshot, ReplicationObjectBridge, replication_state_to_filemeta,
};
use crate::bucket::replication::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
use crate::storage_api_contracts::object::DeletedObject;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) type LifecycleReplicationConfig = ReplicationLifecycleConfig;
pub(crate) fn has_pending_version_purge(obj: &ObjectOpts) -> bool {
obj.version_purge_status.is_pending()
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn has_pending_object_replication(obj: &ObjectOpts) -> bool {
replication_status_blocks_lifecycle(&obj.replication_status)
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn has_pending_lifecycle_replication(obj: &ObjectOpts) -> bool {
has_pending_object_replication(obj) || has_pending_version_purge(obj)
}
@@ -14,10 +14,6 @@
use std::collections::HashMap;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn decode_tags_to_map(tags: &str) -> HashMap<String, String> {
crate::bucket::tagging::decode_tags_to_map(tags)
}
@@ -331,10 +331,6 @@ where
persist_tier_delete_journal_entry(api, &committed).await
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn abort_tier_delete_journal_entry<S>(api: Arc<S>, je: &Jentry) -> std::io::Result<()>
where
S: ObjectOperations<
@@ -148,10 +148,6 @@ struct RecoveryCursor {
object: String,
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn recover_tier_free_versions(
api: Arc<ECStore>,
limit: usize,
@@ -80,10 +80,7 @@ impl LastDayTierStats {
}
}
#[allow(
dead_code,
reason = "asserted by this file's tests; the lib target cannot see test-only consumers (backlog#1823)"
)]
#[allow(dead_code)]
fn merge(&self, m: LastDayTierStats) -> LastDayTierStats {
let mut cl = self.clone();
let mut cm = m;
@@ -177,10 +177,9 @@ fn should_record_remote_delete_failure(err: &std::io::Error) -> bool {
}
#[derive(Default)]
#[allow(dead_code)]
struct ObjSweeper {
#[allow(dead_code, reason = "written but never read back (backlog#1823)")]
object: String,
#[allow(dead_code, reason = "written but never read back (backlog#1823)")]
bucket: String,
version_id: Option<Uuid>,
versioned: bool,
@@ -192,9 +191,9 @@ struct ObjSweeper {
remote_object: String,
}
#[allow(dead_code)]
impl ObjSweeper {
#[allow(clippy::new_ret_no_self)]
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
pub async fn new(bucket: &str, object: &str) -> Result<Self, std::io::Error> {
Ok(Self {
object: object.into(),
@@ -203,20 +202,17 @@ impl ObjSweeper {
})
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
pub fn with_version(&mut self, vid: Option<Uuid>) -> &Self {
self.version_id = vid.clone();
self
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
pub fn with_versioning(&mut self, versioned: bool, suspended: bool) -> &Self {
self.versioned = versioned;
self.suspended = suspended;
self
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
pub fn get_opts(&self) -> lifecycle::ObjectOpts {
let mut opts = ObjectOpts {
version_id: self.version_id.clone(),
@@ -230,7 +226,6 @@ impl ObjSweeper {
opts
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
pub fn set_transition_state(&mut self, info: TransitionedObject) {
self.transition_tier = info.tier;
self.transition_status = info.status;
@@ -271,7 +266,6 @@ impl ObjSweeper {
None
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
pub async fn sweep(&self, api: Arc<ECStore>) {
let Some(je) = self.should_remove_remote_object() else {
return;
@@ -391,10 +385,6 @@ impl ExpiryOp for Jentry {
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> {
let result = delete_object_from_remote_tier_raw(obj_name, rv_id, tier_name).await;
if let Err(err) = &result
@@ -405,10 +395,6 @@ pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_na
result
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
async fn delete_object_from_remote_tier_raw(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> {
#[cfg(test)]
if let Some(result) = run_remote_tier_delete_test_hook(obj_name, rv_id, tier_name) {
@@ -419,10 +405,6 @@ async fn delete_object_from_remote_tier_raw(obj_name: &str, rv_id: &str, tier_na
delete_object_from_remote_tier_raw_with_manager(obj_name, rv_id, tier_name, &tier_config_mgr).await
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
async fn delete_object_from_remote_tier_raw_with_manager(
obj_name: &str,
rv_id: &str,
@@ -503,10 +485,6 @@ pub enum RemoteTierDeleteOutcome {
AlreadyRemoved,
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn delete_object_from_remote_tier_idempotent(
obj_name: &str,
rv_id: &str,
@@ -50,16 +50,8 @@ pub type Result<T> = std::result::Result<T, TransitionTransactionError>;
#[derive(Debug, thiserror::Error)]
pub enum TransitionTransactionError {
#[error("transition transaction already exists")]
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
AlreadyExists,
#[error("transition transaction is not found")]
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
NotFound,
#[error("transition transaction is corrupt: {0}")]
Corrupt(&'static str),
-31
View File
@@ -60,14 +60,12 @@ struct ConfigWriteLockProbeState {
static CONFIG_WRITE_LOCK_PROBES: std::sync::OnceLock<StdMutex<Vec<Arc<ConfigWriteLockProbeState>>>> = std::sync::OnceLock::new();
#[cfg(any(test, feature = "test-util"))]
#[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
pub struct ConfigWriteLockProbe {
state: Arc<ConfigWriteLockProbeState>,
}
#[cfg(any(test, feature = "test-util"))]
impl ConfigWriteLockProbe {
#[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
pub fn install(bucket: &str) -> Self {
let state = Arc::new(ConfigWriteLockProbeState {
bucket: bucket.to_string(),
@@ -86,7 +84,6 @@ impl ConfigWriteLockProbe {
Self { state }
}
#[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
pub async fn wait_until_attempted(&self) {
tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
.await
@@ -659,16 +656,6 @@ pub async fn update_under_transaction_lock(
update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data).await
}
/// Clear one config file while the caller holds this bucket's transaction lock.
pub async fn delete_under_transaction_lock(
guard: &BucketMetadataMutationGuard,
bucket: &str,
config_file: &str,
) -> Result<OffsetDateTime> {
guard.ensure_valid(bucket)?;
delete_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file).await
}
pub async fn update_quota_if_incarnation(
bucket: &str,
data: Vec<u8>,
@@ -808,14 +795,6 @@ pub async fn acquire_bucket_metadata_transaction_lock(bucket: &str) -> Result<Bu
acquire_config_write_guard(get_bucket_metadata_sys()?, bucket).await
}
/// Acquire the bucket transaction lock only if its incarnation still matches.
pub async fn acquire_bucket_metadata_transaction_lock_for_incarnation(
bucket: &str,
expected_incarnation_id: Uuid,
) -> Result<BucketMetadataMutationGuard> {
acquire_config_write_guard_for_incarnation(get_bucket_metadata_sys()?, bucket, Some(expected_incarnation_id)).await
}
pub(crate) async fn acquire_bucket_metadata_transaction_lock_in(
ctx: &crate::runtime::instance::InstanceContext,
bucket: &str,
@@ -893,10 +872,6 @@ pub async fn get_bucket_policy_raw(bucket: &str) -> Result<(String, OffsetDateTi
bucket_meta_sys.get_bucket_policy_raw(bucket).await
}
#[allow(
dead_code,
reason = "free-function facade over the live BucketMetadataSys::get_bucket_acl_config; no caller in this port (backlog#1823)"
)]
pub async fn get_bucket_acl_config(bucket: &str) -> Result<(String, OffsetDateTime)> {
let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -1111,10 +1086,6 @@ pub async fn get_config_from_disk(bucket: &str) -> Result<BucketMetadata> {
bucket_meta_sys.get_config_from_disk(bucket).await
}
#[allow(
dead_code,
reason = "ambient-facade variant of the live created_at_in; no caller in this port (backlog#1823)"
)]
pub async fn created_at(bucket: &str) -> Result<OffsetDateTime> {
let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -1628,7 +1599,6 @@ impl BucketMetadataSys {
/// [`Self::update`], with the payload computed from the loaded metadata
/// instead of supplied up front. Loads through this system's own store so
/// the read and the persisted write target the same instance.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
async fn update_config_with<F>(&self, bucket: &str, config_file: &str, mutate: F) -> Result<OffsetDateTime>
where
F: FnOnce(&BucketMetadata) -> Result<Vec<u8>> + Send,
@@ -1733,7 +1703,6 @@ impl BucketMetadataSys {
/// A miss is never published as an authoritative default, and a snapshot
/// read before delete plus same-name recreation cannot replace the new
/// generation.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(crate) async fn reload_from_store(&self, bucket: &str) -> Result<()> {
if is_meta_bucketname(bucket) {
return Err(Error::other("errInvalidArgument"));
+1
View File
@@ -13,6 +13,7 @@
// limitations under the License.
// #730: bucket subsystems still contain staged ECStore migration code.
#![allow(dead_code)]
pub mod bandwidth;
pub mod bucket_target_sys;
@@ -136,7 +136,6 @@ pub fn add_years(dt: OffsetDateTime, years: i32) -> OffsetDateTime {
/// Check if an object has legal hold enabled.
/// Returns true if legal hold is ON.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> bool {
let lhold = objectlock::get_object_legalhold_meta(user_defined);
matches!(lhold.status, Some(ref st) if st.as_str() == ObjectLockLegalHoldStatus::ON)
@@ -152,7 +151,6 @@ fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> b
/// # Returns
/// * `true` if the object is locked (cannot be deleted/modified)
/// * `false` if the object is not locked
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn is_object_locked_by_metadata(user_defined: &std::collections::HashMap<String, String>, is_delete_marker: bool) -> bool {
// Delete markers are never locked
if is_delete_marker {
+2 -2
View File
@@ -193,7 +193,6 @@ pub enum QuotaError {
}
#[derive(Debug, Serialize)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub struct QuotaErrorResponse {
#[serde(rename = "Code")]
pub code: String,
@@ -209,7 +208,6 @@ pub struct QuotaErrorResponse {
}
impl QuotaErrorResponse {
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn new(quota_error: &QuotaError, request_id: &str, host_id: &str) -> Self {
match quota_error {
QuotaError::QuotaExceeded { .. } => Self {
@@ -312,7 +310,9 @@ mod tests {
}
#[derive(Deserialize)]
struct LegacyBucketQuota {
#[allow(dead_code)]
quota: Option<u64>,
#[allow(dead_code)]
quota_type: LegacyQuotaType,
}
let legacy = serde_json::from_slice::<LegacyBucketQuota>(&json)
@@ -899,7 +899,6 @@ async fn save_ledger_locked(
}
#[cfg(any(test, feature = "test-util"))]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn fail_next_quota_ledger_save_for_test() {
FAIL_NEXT_LEDGER_SAVE.store(true, std::sync::atomic::Ordering::SeqCst);
}
+2 -2
View File
@@ -60,7 +60,7 @@ pub use replication_filemeta_boundary::{
pub(crate) use replication_filemeta_boundary::{
replication_state_from_filemeta, replication_status_from_filemeta, version_purge_status_from_filemeta,
};
pub(crate) use replication_lifecycle_bridge::ReplicationLifecycleBridge;
pub(crate) use replication_lifecycle_bridge::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
pub(crate) use replication_migration_bridge::ReplicationMigrationBridge;
pub use replication_object_bridge::ReplicationObjectBridge;
pub use replication_object_config::{DeleteReplicationConfigSnapshot, ReplicationConfig};
@@ -81,6 +81,6 @@ pub use replication_queue_boundary::{
pub use replication_resync_boundary::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus};
pub use replication_scanner_bridge::ReplicationScannerBridge;
pub use replication_state::{ReplicationStats, RuntimeReplicationTargetBacklog};
pub use replication_stats_boundary::{BucketReplicationStat, BucketReplicationStats, BucketStats, InQueueMetric, XferStats};
pub use replication_stats_boundary::{BucketReplicationStats, BucketStats};
pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage};
pub(crate) use replication_target_config_bridge::ReplicationTargetConfigBridge;
@@ -37,10 +37,6 @@ impl ReplicationConfigStore {
com::read_config_limited(api, file, max_bytes).await
}
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub(crate) async fn read_no_lock<S>(api: Arc<S>, file: &str) -> Result<Vec<u8>>
where
S: ReplicationObjectIO,
@@ -24,27 +24,15 @@ use super::replication_storage_boundary::{
DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete, deleted_object_for_replication,
};
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) type ReplicationLifecycleConfig = ReplicationConfig;
pub(crate) struct ReplicationLifecycleBridge;
impl ReplicationLifecycleBridge {
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn new_config(config: ReplicationConfiguration) -> ReplicationLifecycleConfig {
ReplicationConfig::new(Some(config), None)
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn has_pending_version_purge(
config: &ReplicationLifecycleConfig,
object_name: &str,
@@ -57,10 +45,6 @@ impl ReplicationLifecycleBridge {
.is_some_and(|config| config.has_active_rules(object_name, true))
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) async fn check_delete_replication(
bucket: &str,
object: &ObjectToDelete,
@@ -70,10 +54,6 @@ impl ReplicationLifecycleBridge {
check_replicate_delete(bucket, object, source, opts, None).await
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn version_delete_replication_state(decision: &ReplicateDecision) -> ReplicationState {
let pending_status = decision.pending_status();
ReplicationState {
@@ -19,33 +19,17 @@ use time::OffsetDateTime;
use super::replication_error_boundary::Result;
use crate::bucket::msgp_decode;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) struct ReplicationMsgpCodec;
impl ReplicationMsgpCodec {
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn read_ext8_time<R: Read>(rd: &mut R) -> Result<OffsetDateTime> {
msgp_decode::read_msgp_ext8_time(rd)
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn skip_value<R: Read>(rd: &mut R) -> Result<()> {
msgp_decode::skip_msgp_value(rd)
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn write_time<W: Write>(wr: &mut W, time: OffsetDateTime) -> Result<()> {
msgp_decode::write_msgp_time(wr, time)
}
@@ -77,10 +77,6 @@ impl ReplicationObjectBridge {
load_delete_request_config_in(ctx, bucket).await
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) async fn delete_config_snapshot_in(
ctx: &ReplicationInstanceContext,
bucket: &str,
@@ -231,10 +231,6 @@ pub(crate) async fn load_delete_replication_config(
delete_snapshot_from_metadata(ReplicationMetadataStore::delete_metadata(bucket).await?)
}
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub(crate) async fn load_delete_replication_config_in(
ctx: &ReplicationInstanceContext,
bucket: &str,
@@ -217,10 +217,6 @@ impl DurableMrfBacklogTracker {
}
}
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
fn durable_mrf_backlog_tracker_from_entries(entries: &[MrfReplicateEntry]) -> DurableMrfBacklogTracker {
let mut tracker = DurableMrfBacklogTracker {
available: true,
@@ -716,10 +712,6 @@ pub struct ReplicationPool<S: ReplicationStorage> {
// MRF worker lifecycle
mrf_worker_cancellations: Mutex<Vec<CancellationToken>>,
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
mrf_stop_tx: Sender<()>,
// Worker size tracking
@@ -948,10 +940,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Resizes worker priority and counts
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub async fn resize_worker_priority(
&self,
pri: ReplicationPriority,
@@ -1192,10 +1180,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Queues an MRF save operation
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
async fn queue_mrf_save(&self, entry: MrfReplicateEntry) {
let _ = self.queue_mrf_save_admission(entry, "mrf_worker").await;
}
@@ -1667,10 +1651,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Worker function for handling regular replication operations
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
async fn add_worker(
&self,
mut rx: Receiver<ReplicationOperation>,
@@ -1684,10 +1664,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Worker function for handling large object replication operations
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
async fn add_large_worker(
&self,
mut rx: Receiver<ReplicationOperation>,
@@ -1702,10 +1678,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Worker function for handling MRF (Most Recent Failures) operations
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
async fn add_mrf_worker(
&self,
mut rx: Receiver<ReplicationOperation>,
@@ -1719,10 +1691,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Delete resync metadata from replication resync state in memory
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub async fn delete_resync_metadata(&self, bucket: &str) {
let mut status_map = self.resyncer.status_map.write().await;
status_map.remove(bucket);
@@ -21,31 +21,11 @@ pub(crate) use rustfs_replication::{
should_count_head_proxy_failure,
};
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const RESYNC_META_FORMAT: u16 = rustfs_replication::resync::RESYNC_META_FORMAT;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const RESYNC_META_VERSION: u16 = rustfs_replication::resync::RESYNC_META_VERSION;
pub(crate) const RESYNC_FILE_MAX_BYTES: usize = rustfs_replication::RESYNC_FILE_MAX_BYTES;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const WIRE_ZERO_TIME_UNIX: i64 = rustfs_replication::resync::WIRE_ZERO_TIME_UNIX;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const MRF_META_FORMAT: u16 = rustfs_replication::mrf::MRF_META_FORMAT;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const MRF_META_VERSION: u16 = rustfs_replication::mrf::MRF_META_VERSION;
fn map_replication_error(err: rustfs_replication::Error) -> Error {
@@ -122,10 +122,6 @@ const REPLICATION_TARGET_OFFLINE_ERROR_MARKERS: &[&str] = &[
"tcp connect error",
];
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
const RESYNC_TIME_INTERVAL: TokioDuration = TokioDuration::from_secs(60);
static WARNED_MONITOR_UNINIT: std::sync::Once = std::sync::Once::new();
@@ -332,10 +328,6 @@ fn bounded_resync_max_jobs(value: usize) -> usize {
#[derive(Debug)]
pub struct ReplicationResyncer {
pub status_map: Arc<RwLock<HashMap<String, BucketReplicationResyncStatus>>>,
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub worker_size: usize,
pub(crate) cancel_tokens: Arc<RwLock<HashMap<ResyncCancelKey, CancellationToken>>>,
resync_admission: Arc<Semaphore>,
@@ -552,10 +544,6 @@ impl ReplicationResyncer {
.is_some_and(|status| status.failed_count > 0)
}
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub async fn persist_to_disk<S>(&self, cancel_token: CancellationToken, api: Arc<S>)
where
S: ReplicationObjectIO,
@@ -340,10 +340,6 @@ impl ReplicationStats {
}
/// Site replication update replica statistics
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
fn sr_update_replica_stat(&self, size: i64) {
self.sr_stats.replica_size.fetch_add(size, Ordering::Relaxed);
self.sr_stats.replica_count.fetch_add(1, Ordering::Relaxed);
@@ -708,12 +704,6 @@ impl ReplicationStats {
} else {
BucketReplicationStats::new()
};
// Stamp the serializable failure windows from the live samples: the
// samples themselves do not cross the peer-RPC wire, so this snapshot
// is what cluster aggregation and the metrics endpoints see.
for stat in replication_stats.stats.values_mut() {
stat.fail_stats.refresh_windows();
}
let uptime = if cache.contains_key(bucket) {
SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
@@ -15,9 +15,7 @@
#[cfg(test)]
pub(crate) use rustfs_replication::FailStats;
pub(crate) use rustfs_replication::{
ActiveWorkerStat, ProxyMetric, ProxyStatsCache, QueueCache, ReplicationMetricScope, SRMetricsSummary,
ActiveWorkerStat, BucketReplicationStat, InQueueMetric, ProxyMetric, ProxyStatsCache, QueueCache, ReplicationMetricScope,
SRMetricsSummary, XferStats,
};
// Public so the admin wire DTOs (rustfs/src/admin/replication_metrics_wire.rs)
// can project the internal stats onto the minio-go response shapes through
// the storage_api facade chain.
pub use rustfs_replication::{BucketReplicationStat, BucketReplicationStats, BucketStats, InQueueMetric, XferStats};
pub use rustfs_replication::{BucketReplicationStats, BucketStats};
@@ -27,10 +27,8 @@ use rustfs_utils::http::{
AMZ_OBJECT_TAGGING, AMZ_SERVER_SIDE_ENCRYPTION, AMZ_SERVER_SIDE_ENCRYPTION_KMS_CONTEXT, AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID,
AMZ_STORAGE_CLASS, AMZ_TAG_COUNT, CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE,
HeaderExt as _, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP,
SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP,
SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_TAGGING_TIMESTAMP,
get_str, insert_header_map, is_internal_key, is_object_encryption_marker, is_replication_stripped_encryption_key,
ssec_replication_transport_header,
SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_TAGGING_TIMESTAMP, get_str, insert_header_map,
is_internal_key, is_object_encryption_marker, is_replication_stripped_encryption_key, ssec_replication_transport_header,
};
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
@@ -121,27 +119,6 @@ fn classify_replication_source_encryption(metadata: &HashMap<String, String>) ->
}
}
fn is_legacy_source_replication_timestamp_key(key: &str) -> bool {
fn has_prefix_and_suffix(key: &str, prefix: &str, suffix: &str) -> bool {
let key = key.as_bytes();
key.len() == prefix.len() + suffix.len()
&& key[..prefix.len()].eq_ignore_ascii_case(prefix.as_bytes())
&& key[prefix.len()..].eq_ignore_ascii_case(suffix.as_bytes())
}
[
SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP,
SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP,
SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP,
]
.iter()
.any(|suffix| {
["x-rustfs-", "x-minio-"]
.iter()
.any(|prefix| has_prefix_and_suffix(key, prefix, suffix))
})
}
pub(crate) fn replication_object_is_ssec_encrypted(user_defined: &HashMap<String, String>) -> bool {
rustfs_replication::is_ssec_encrypted(user_defined)
}
@@ -199,11 +176,6 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
continue;
}
if is_legacy_source_replication_timestamp_key(key) {
meta.insert(format!("x-amz-meta-{key}"), value.to_string());
continue;
}
if is_internal_key(key) || is_standard_header(key) {
continue;
}
@@ -287,23 +259,15 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
if !tags.is_empty() {
put_options.user_tags = tags;
put_options.internal.tagging_timestamp =
if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP) {
OffsetDateTime::parse(&timestamp, &Rfc3339)
.map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))?
} else {
object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
};
}
}
// Load the stored tagging timestamp independently of whether any tags
// remain: DeleteObjectTagging leaves the object tagless but stamps this
// key, and the deletion's LWW timestamp must still reach the replica.
// With no stored key, fall back to mod_time only while tags exist
// (MinIO parity); a tagless object without the key was never tagged and
// keeps the epoch default (no header).
put_options.internal.tagging_timestamp = if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP)
{
OffsetDateTime::parse(&timestamp, &Rfc3339)
.map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))?
} else if !put_options.user_tags.is_empty() {
object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
} else {
OffsetDateTime::UNIX_EPOCH
};
let metadata = &*object_info.user_defined;
@@ -319,15 +283,13 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
put_options.cache_control = cache_control.to_string();
}
if let Some(mode) = metadata.lookup(AMZ_OBJECT_LOCK_MODE).filter(|mode| !mode.is_empty()) {
if let Some(mode) = metadata.lookup(AMZ_OBJECT_LOCK_MODE) {
put_options.mode = Some(ObjectLockRetentionMode::from(mode.to_uppercase().as_str()));
}
if let Some(retain_until_date) = metadata.lookup(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE) {
if !retain_until_date.is_empty() {
put_options.retain_until_date = OffsetDateTime::parse(retain_until_date, &Rfc3339)
.map_err(|err| Error::other(format!("Failed to parse retain until date: {err}")))?;
}
put_options.retain_until_date = OffsetDateTime::parse(retain_until_date, &Rfc3339)
.map_err(|err| Error::other(format!("Failed to parse retain until date: {err}")))?;
put_options.internal.retention_timestamp =
if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP) {
OffsetDateTime::parse(&timestamp, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH)
@@ -732,110 +694,6 @@ mod tests {
assert!(options.internal.replication_request);
}
/// DeleteObjectTagging leaves the object tagless but stamps the
/// tagging-timestamp internal key; the deletion's LWW timestamp must
/// still be loaded (and therefore sent) so the replica can order the
/// deletion against concurrent tag edits.
#[test]
fn replication_put_options_carry_tagging_timestamp_after_tag_deletion() {
let mut metadata = std::collections::HashMap::new();
rustfs_utils::http::insert_str(&mut metadata, SUFFIX_TAGGING_TIMESTAMP, "2026-01-02T03:04:05Z".to_string());
let object_info = ObjectInfo {
user_defined: Arc::new(metadata),
user_tags: Arc::new(String::new()),
mod_time: Some(OffsetDateTime::UNIX_EPOCH),
version_id: Some(Uuid::nil()),
..Default::default()
};
let (options, _) = replication_put_object_options("", &object_info).expect("build put options");
assert!(options.user_tags.is_empty());
assert_eq!(
options.internal.tagging_timestamp,
OffsetDateTime::parse("2026-01-02T03:04:05Z", &Rfc3339).expect("valid timestamp"),
"the stored tagging timestamp must load independently of remaining tags"
);
// A tagless object without the stored key was never tagged: the epoch
// default keeps the header unsent.
let untagged = ObjectInfo {
user_tags: Arc::new(String::new()),
mod_time: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")),
version_id: Some(Uuid::nil()),
..Default::default()
};
let (options, _) = replication_put_object_options("", &untagged).expect("build put options");
assert_eq!(options.internal.tagging_timestamp, OffsetDateTime::UNIX_EPOCH);
}
#[test]
fn replication_put_options_do_not_promote_legacy_user_timestamp_metadata() {
let legacy_keys = [
"x-rustfs-source-replication-tagging-timestamp",
"x-rustfs-source-replication-retention-timestamp",
"x-rustfs-source-replication-legalhold-timestamp",
"x-minio-source-replication-tagging-timestamp",
"x-minio-source-replication-retention-timestamp",
"x-minio-source-replication-legalhold-timestamp",
];
let object_info = ObjectInfo {
user_defined: Arc::new(
legacy_keys
.iter()
.map(|key| (key.to_string(), "2099-01-02T03:04:05Z".to_string()))
.collect(),
),
..Default::default()
};
let (options, _) = replication_put_object_options("", &object_info).expect("build put options");
for legacy_key in legacy_keys {
assert!(!options.user_metadata.contains_key(legacy_key));
assert_eq!(
options
.user_metadata
.get(&format!("x-amz-meta-{legacy_key}"))
.map(String::as_str),
Some("2099-01-02T03:04:05Z")
);
}
assert_eq!(options.internal.tagging_timestamp, OffsetDateTime::UNIX_EPOCH);
assert_eq!(options.internal.retention_timestamp, OffsetDateTime::UNIX_EPOCH);
assert_eq!(options.internal.legalhold_timestamp, OffsetDateTime::UNIX_EPOCH);
}
#[test]
fn replication_put_options_carry_retention_timestamp_after_clear() {
let mut metadata = HashMap::from([
(AMZ_OBJECT_LOCK_MODE.to_string(), String::new()),
(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.to_string(), String::new()),
]);
rustfs_utils::http::insert_str(&mut metadata, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, "2026-01-02T03:04:05Z".to_string());
let object_info = ObjectInfo {
user_defined: Arc::new(metadata),
..Default::default()
};
let (options, _) = replication_put_object_options("", &object_info).expect("retention clear must replicate");
assert!(options.mode.is_none());
assert_eq!(options.retain_until_date, OffsetDateTime::UNIX_EPOCH);
assert_eq!(
options.internal.retention_timestamp,
OffsetDateTime::parse("2026-01-02T03:04:05Z", &Rfc3339).expect("valid timestamp")
);
let headers = options.header();
assert!(!headers.contains_key(AMZ_OBJECT_LOCK_MODE));
assert!(!headers.contains_key(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE));
assert_eq!(
rustfs_utils::http::get_header(&headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP).as_deref(),
Some("2026-01-02T03:04:05Z")
);
}
#[test]
fn replication_put_options_strip_encryption_metadata_from_plaintext_objects() {
use rustfs_utils::http::object_encryption_keys::{INTERNAL_ENCRYPTION_ORIGINAL_SIZE_HEADER, SSEC_ORIGINAL_SIZE_HEADER};
+4 -52
View File
@@ -40,14 +40,7 @@ impl ARN {
impl Display for ARN {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
// The `minio` partition is deliberate: madmin-go's ParseARN
// hard-rejects any other partition, so native mc/madmin tooling can
// only decode remote-target ARNs minted in this form (backlog#1675
// P1-7). Legacy `arn:rustfs:` ARNs persisted by older releases stay
// readable via the FromStr whitelist below; runtime matching between
// targets and replication rules is by full-string equality, so mixed
// partitions coexist safely.
write!(f, "arn:minio:{}:{}:{}:{}", self.arn_type, self.region, self.id, self.bucket)
write!(f, "arn:rustfs:{}:{}:{}:{}", self.arn_type, self.region, self.id, self.bucket)
}
}
@@ -55,12 +48,7 @@ impl FromStr for ARN {
type Err = std::io::Error;
fn from_str(s: &str) -> Result<Self, Self::Err> {
// Partition whitelist, not just an `arn:` check: `BucketTargetType::
// from_str(...).unwrap_or_default()` below never fails, so this is
// the only structural gate rejecting foreign ARNs. `arn:rustfs:` is
// the legacy partition and must stay accepted forever (persisted
// bucket-targets.json / replication configs from older releases).
if !s.starts_with("arn:minio:") && !s.starts_with("arn:rustfs:") {
if !s.starts_with("arn:rustfs:") {
return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, "Invalid ARN format"));
}
@@ -113,50 +101,14 @@ mod tests {
}
/// RustFS commonly generates ARNs with an empty region:
/// `arn:minio:replication::<deployment_id>:<bucket>`.
/// `arn:rustfs:replication::<deployment_id>:<bucket>`.
#[test]
fn from_str_handles_empty_region_segment() {
let parsed = ARN::from_str("arn:minio:replication::depl-123:bucket-a").expect("valid ARN must parse");
let parsed = ARN::from_str("arn:rustfs:replication::depl-123:bucket-a").expect("valid ARN must parse");
assert_eq!(parsed.arn_type, BucketTargetType::ReplicationService);
assert_eq!(parsed.region, "", "region segment is empty in this form");
assert_eq!(parsed.id, "depl-123");
assert_eq!(parsed.bucket, "bucket-a");
}
/// madmin-go's `ParseARN` hard-rejects anything that does not start with
/// `arn:minio:`, so generated ARNs must use the `minio` partition or the
/// native mc/madmin tooling cannot decode remote-target listings.
#[test]
fn display_emits_minio_partition() {
let arn = ARN::new(
BucketTargetType::ReplicationService,
"depl-123".to_string(),
String::new(),
"bucket-a".to_string(),
);
assert_eq!(arn.to_string(), "arn:minio:replication::depl-123:bucket-a");
}
/// Persisted bucket-targets.json files from older RustFS releases carry
/// `arn:rustfs:` ARNs; the legacy partition must stay parseable forever.
#[test]
fn from_str_accepts_legacy_rustfs_partition() {
let parsed = ARN::from_str("arn:rustfs:replication:us-east-1:depl-123:bucket-a").expect("legacy ARN must parse");
assert_eq!(parsed.arn_type, BucketTargetType::ReplicationService);
assert_eq!(parsed.region, "us-east-1");
assert_eq!(parsed.id, "depl-123");
assert_eq!(parsed.bucket, "bucket-a");
}
/// The partition whitelist is the only structural gate: `BucketTargetType::
/// from_str(...).unwrap_or_default()` never fails, so any 6-segment string
/// would otherwise parse as `type=None`.
#[test]
fn from_str_rejects_unknown_partition() {
assert!(ARN::from_str("arn:aws:replication::depl-123:bucket-a").is_err());
assert!(ARN::from_str("not-an-arn").is_err());
}
}
@@ -59,10 +59,6 @@ impl fmt::Debug for Credentials {
}
#[derive(Debug, Deserialize, Serialize, Default, Clone)]
#[allow(
dead_code,
reason = "MinIO-parity bucket-target service discriminator with no caller in this port (backlog#1823)"
)]
pub enum ServiceType {
#[default]
Replication,
+17 -20
View File
@@ -73,6 +73,23 @@ pub fn check_valid_bucket_name_strict(bucket_name: &str) -> Result<()> {
check_bucket_name_common(bucket_name, true)
}
pub fn check_valid_object_name_prefix(object_name: &str) -> Result<()> {
if object_name.len() > 1024 {
return Err(Error::other("Object name cannot be longer than 1024 characters"));
}
if !object_name.is_ascii() {
return Err(Error::other("Object name with non-UTF-8 strings are not supported"));
}
Ok(())
}
pub fn check_valid_object_name(object_name: &str) -> Result<()> {
if object_name.trim().is_empty() {
return Err(Error::other("Object name cannot be empty"));
}
check_valid_object_name_prefix(object_name)
}
pub fn deserialize<T>(input: &[u8]) -> xml::DeResult<T>
where
T: for<'xml> xml::Deserialize<'xml>,
@@ -83,10 +100,6 @@ where
Ok(ans)
}
#[allow(
dead_code,
reason = "xml serialize helper with no caller in this port; the live sibling is deserialize (backlog#1823)"
)]
pub fn serialize_content<T: xml::SerializeContent>(val: &T) -> xml::SerResult<String> {
let mut buf = Vec::with_capacity(256);
{
@@ -173,27 +186,15 @@ pub fn is_valid_object_name(object: &str) -> bool {
/// Client-facing reason attached to rejections of object keys that Win32/NTFS
/// cannot represent as file paths (issue #3299). Deployments on Linux/macOS
/// accept the full S3 key character set.
#[allow(
dead_code,
reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
)]
pub const WINDOWS_RESERVED_CHARACTERS_REASON: &str =
"object key contains characters unsupported on Windows hosts (one of ':', '*', '?', '\"', '|', '<', '>')";
/// Client-facing reason for path segments Windows can store but not address
/// afterwards (issue #3449): trailing dot/space or reserved DOS device names.
#[allow(
dead_code,
reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
)]
pub const WINDOWS_RESERVED_SEGMENT_REASON: &str = "object key contains a path segment unsupported on Windows hosts (trailing dot or space, or a reserved device name such as NUL/CON/COM1)";
/// Reserved DOS device names that shadow regular files on Windows, even when
/// an extension is appended (e.g. `NUL.txt` resolves to the `NUL` device).
#[allow(
dead_code,
reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
)]
const WINDOWS_RESERVED_NAMES: &[&str] = &[
"CON", "PRN", "AUX", "NUL", "COM1", "COM2", "COM3", "COM4", "COM5", "COM6", "COM7", "COM8", "COM9", "LPT1", "LPT2", "LPT3",
"LPT4", "LPT5", "LPT6", "LPT7", "LPT8", "LPT9",
@@ -203,10 +204,6 @@ const WINDOWS_RESERVED_NAMES: &[&str] = &[
/// the Win32 API cannot address afterwards (issue #3449): segments ending in a
/// dot or a space, and reserved DOS device names — bare or with an extension
/// (`NUL.txt`), matching classic Win32 path resolution semantics.
#[allow(
dead_code,
reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
)]
pub fn object_name_has_windows_incompatible_segment(object: &str) -> bool {
object.split(['/', '\\']).any(|segment| {
if segment.ends_with('.') || segment.ends_with(' ') {
@@ -90,10 +90,6 @@ impl BucketVersioningSys {
/// caller's own instance context so a second in-process store never
/// answers with the first instance's versioning state; falls back to the
/// ambient system when the instance cell is not initialized.
#[allow(
dead_code,
reason = "instance-scoped seam (backlog#1052) with no caller in this port (backlog#1823)"
)]
pub(crate) async fn get_in(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result<VersioningConfiguration> {
if bucket == RUSTFS_META_BUCKET || bucket.starts_with(RUSTFS_META_BUCKET) {
return Ok(VersioningConfiguration::default());
@@ -15,7 +15,6 @@
use crate::disk::disk_store::{get_drive_walkdir_peek_timeout, get_drive_walkdir_stall_timeout};
use crate::disk::error::DiskError;
use crate::disk::{self, DiskAPI, DiskStore, WalkDirOptions};
use futures::future::join_all;
use metrics::counter;
use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetacacheReader, is_io_eof};
use std::{
@@ -656,7 +655,6 @@ async fn list_path_raw_inner(
errs.push(None);
}
let mut pending_entries: Vec<Option<MetaCacheEntry>> = vec![None; readers.len()];
let mut peek_outcomes: Vec<Option<PeekOutcome>> = std::iter::repeat_with(|| None).take(readers.len()).collect();
loop {
let mut current = MetaCacheEntry::default();
@@ -678,21 +676,6 @@ async fn list_path_raw_inner(
let mut has_err = 0;
let mut agree = 0;
// Start every missing head read in the same round so one stalled
// disk cannot multiply the wait budget by the erasure-set width.
// Outcomes are still consumed below in stable disk-index order.
let concurrent_peeks = readers.iter_mut().enumerate().filter_map(|(i, reader)| {
if errs[i].is_some() || pending_entries[i].is_some() {
return None;
}
let cancel = &revjob_rx;
Some(async move { (i, peek_with_timeout(cancel, reader, peek_timeout).await) })
});
for (i, outcome) in join_all(concurrent_peeks).await {
peek_outcomes[i] = Some(outcome);
}
for (i, r) in readers.iter_mut().enumerate() {
if errs[i].is_some() {
has_err += 1;
@@ -702,10 +685,7 @@ async fn list_path_raw_inner(
let entry = if let Some(entry) = pending_entries[i].take() {
entry
} else {
let Some(outcome) = peek_outcomes[i].take() else {
return Err(DiskError::Unexpected);
};
match outcome {
match peek_with_timeout(&revjob_rx, r, peek_timeout).await {
PeekOutcome::Ready(res) => {
if let Some(entry) = res {
// info!("read entry disk: {}, name: {}", i, entry.name);
@@ -1315,36 +1295,6 @@ mod tests {
assert_eq!(err, DiskError::Timeout);
}
#[tokio::test(start_paused = true)]
async fn list_path_raw_bounds_multiple_stalled_readers_by_one_peek_deadline() {
let peek_timeout = Duration::from_millis(20);
let started = tokio::time::Instant::now();
let err = list_path_raw(
CancellationToken::new(),
ListPathRawOptions {
disks: vec![None, None, None, None],
min_disks: 1,
test_reader_behaviors: vec![
TestReaderBehavior::Stall,
TestReaderBehavior::Stall,
TestReaderBehavior::Stall,
TestReaderBehavior::Stall,
],
peek_timeout: Some(peek_timeout),
..Default::default()
},
)
.await
.expect_err("all stalled readers should fail the listing");
assert_eq!(err, DiskError::Timeout);
assert_eq!(
started.elapsed(),
peek_timeout,
"reader deadlines must overlap instead of accumulating once per disk"
);
}
#[tokio::test]
async fn list_path_raw_waits_past_producer_stall_for_slow_progressing_reader() {
let entry = MetaCacheEntry {
@@ -229,6 +229,17 @@ pub fn http_resp_to_error_response(
err_resp
}
pub fn err_transfer_acceleration_bucket(bucket_name: &str) -> ErrorResponse {
ErrorResponse {
status_code: StatusCode::BAD_REQUEST,
code: S3ErrorCode::InvalidArgument,
message: "The name of the bucket used for Transfer Acceleration must be DNS-compliant and must not contain periods .."
.to_string(),
bucket_name: bucket_name.to_string(),
..Default::default()
}
}
pub fn err_entity_too_large(total_size: i64, max_object_size: i64, bucket_name: &str, object_name: &str) -> ErrorResponse {
let msg = format!(
"Your proposed upload size {} exceeds the maximum allowed object size {} for single PUT operation.",
@@ -284,6 +295,16 @@ pub fn err_invalid_argument(message: &str) -> ErrorResponse {
}
}
pub fn err_api_not_supported(message: &str) -> ErrorResponse {
ErrorResponse {
status_code: StatusCode::NOT_IMPLEMENTED,
code: S3ErrorCode::Custom("APINotSupported".into()),
message: message.to_string(),
request_id: "rustfs".to_string(),
..Default::default()
}
}
#[cfg(test)]
mod tests {
use super::*;
+2 -30
View File
@@ -95,6 +95,7 @@ impl TransitionClient {
}
#[derive(Default)]
#[allow(dead_code)]
pub struct GetRequest {
pub buffer: Vec<u8>,
pub offset: i64,
@@ -106,12 +107,11 @@ pub struct GetRequest {
pub setting_object_info: bool,
}
#[allow(dead_code)]
pub struct GetResponse {
pub size: i64,
//pub error: error,
#[allow(dead_code, reason = "written but never read back (backlog#1823)")]
pub did_read: bool,
#[allow(dead_code, reason = "written but never read back (backlog#1823)")]
pub object_info: ObjectInfo,
}
@@ -135,10 +135,6 @@ impl Object {
Self { ..Default::default() }
}
#[allow(
dead_code,
reason = "MinIO-parity reader surface with no caller in this port (backlog#1823)"
)]
fn do_get_request(&self, request: &GetRequest) -> Result<GetResponse, std::io::Error> {
let _ = request.did_offset_change;
let _ = request.offset;
@@ -154,20 +150,12 @@ impl Object {
))
}
#[allow(
dead_code,
reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
)]
fn set_offset(&mut self, bytes_read: i64) -> Result<(), std::io::Error> {
self.curr_offset += bytes_read;
Ok(())
}
#[allow(
dead_code,
reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
)]
fn read(&mut self, b: &[u8]) -> Result<i64, std::io::Error> {
let mut read_req = GetRequest {
is_read_op: true,
@@ -192,10 +180,6 @@ impl Object {
Ok(response.size)
}
#[allow(
dead_code,
reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
)]
fn stat(&self) -> Result<ObjectInfo, std::io::Error> {
if !self.is_started || !self.object_info_set {
let _ = self.do_get_request(&GetRequest {
@@ -208,10 +192,6 @@ impl Object {
Ok(self.object_info.clone())
}
#[allow(
dead_code,
reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
)]
fn read_at(&mut self, b: &[u8], offset: i64) -> Result<i64, std::io::Error> {
self.curr_offset = offset;
@@ -239,10 +219,6 @@ impl Object {
Ok(response.size)
}
#[allow(
dead_code,
reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
)]
fn seek(&mut self, offset: i64, whence: i64) -> Result<i64, std::io::Error> {
if !self.is_started || !self.object_info_set {
let seek_req = GetRequest {
@@ -277,10 +253,6 @@ impl Object {
Ok(self.curr_offset)
}
#[allow(
dead_code,
reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
)]
fn close(&mut self) -> Result<(), std::io::Error> {
self.is_closed = true;
Ok(())
@@ -27,6 +27,7 @@ use tracing::warn;
use crate::client::api_error_response::err_invalid_argument;
#[derive(Default)]
#[allow(dead_code)]
pub struct AdvancedGetOptions {
pub replication_delete_marker: bool,
pub is_replication_ready_for_delete_marker: bool,
+1
View File
@@ -360,6 +360,7 @@ impl TransitionClient {
}
#[derive(Default)]
#[allow(dead_code)]
pub struct ListObjectsOptions {
reverse_versions: bool,
with_versions: bool,
+2 -4
View File
@@ -37,7 +37,7 @@ use crate::client::{
api_put_object_common::optimal_part_info,
api_put_object_multipart::UploadPartParams,
api_s3_datatypes::{CompleteMultipartUpload, CompletePart, ObjectPart},
constants::{ISO8601_DATEFORMAT, MAX_MULTIPART_PUT_OBJECT_SIZE, MIN_PART_SIZE},
constants::{ISO8601_DATEFORMAT, MAX_MULTIPART_PUT_OBJECT_SIZE, MIN_PART_SIZE, TOTAL_WORKERS},
credentials::SignatureType,
transition_api::{ReaderImpl, TransitionClient, UploadInfo},
utils::{is_amz_header, is_minio_header, is_rustfs_header, is_standard_header, is_storageclass_header},
@@ -137,8 +137,8 @@ impl Default for PutObjectOptions {
}
}
#[allow(dead_code)]
impl PutObjectOptions {
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
fn set_match_etag(&mut self, etag: &str) {
if etag == "*" {
self.custom_header.insert("If-Match", HeaderValue::from_static("*"));
@@ -149,7 +149,6 @@ impl PutObjectOptions {
}
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
fn set_match_etag_except(&mut self, etag: &str) {
if etag == "*" {
self.custom_header.insert("If-None-Match", HeaderValue::from_static("*"));
@@ -260,7 +259,6 @@ impl PutObjectOptions {
header
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
fn validate(&self, c: TransitionClient) -> Result<(), std::io::Error> {
//if self.checksum.is_set() {
/*if !self.trailing_header_support {
@@ -30,6 +30,10 @@ pub fn is_object(reader: &ReaderImpl) -> bool {
matches!(reader, ReaderImpl::ObjectBody(_))
}
pub fn is_read_at(reader: ReaderImpl) -> bool {
matches!(reader, ReaderImpl::ObjectBody(_))
}
pub fn optimal_part_info(object_size: i64, configured_part_size: u64) -> Result<(i64, i64, i64), std::io::Error> {
let unknown_size;
let mut object_size = object_size;
@@ -81,6 +81,18 @@ async fn read_multipart_part(reader: &mut ReaderImpl, want: usize) -> Result<Vec
}
}
pub struct UploadedPartRes {
pub error: std::io::Error,
pub part_num: i64,
pub size: i64,
pub part: ObjectPart,
}
pub struct UploadPartReq {
pub part_num: i64,
pub part: ObjectPart,
}
impl TransitionClient {
pub async fn put_object_multipart_stream(
self: Arc<Self>,
+3 -2
View File
@@ -55,6 +55,7 @@ pub struct RemoveBucketOptions {
const DELETE_RESPONSE_PREVIEW_LEN: usize = 1024;
#[derive(Debug)]
#[allow(dead_code)]
pub struct AdvancedRemoveOptions {
pub replication_delete_marker: bool,
pub replication_status: ReplicationStatus,
@@ -464,10 +465,10 @@ impl TransitionClient {
}
#[derive(Debug, Default)]
#[allow(dead_code)]
pub struct RemoveObjectError {
#[allow(dead_code, reason = "written but never read back (backlog#1823)")]
object_name: String,
#[allow(dead_code, reason = "written but never read back (backlog#1823)")]
#[allow(dead_code)]
version_id: String,
err: Option<std::io::Error>,
}
+38 -18
View File
@@ -29,6 +29,10 @@ use crate::client::utils::base64_decode;
use super::transition_api;
pub struct ListAllMyBucketsResult {
pub owner: Owner,
}
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct CommonPrefix {
pub prefix: String,
@@ -85,10 +89,6 @@ pub struct ListVersionsResult {
pub next_version_id_marker: String,
}
#[allow(
dead_code,
reason = "fields of a MinIO-parity list result that this port builds but never reads back (backlog#1823)"
)]
pub struct ListBucketResult {
common_prefixes: Vec<CommonPrefix>,
contents: Vec<transition_api::ObjectInfo>,
@@ -102,10 +102,6 @@ pub struct ListBucketResult {
prefix: String,
}
#[allow(
dead_code,
reason = "fields of a MinIO-parity list result that this port builds but never reads back (backlog#1823)"
)]
pub struct ListMultipartUploadsResult {
bucket: String,
key_marker: String,
@@ -121,15 +117,16 @@ pub struct ListMultipartUploadsResult {
common_prefixes: Vec<CommonPrefix>,
}
#[allow(
dead_code,
reason = "fields of a MinIO-parity list result that this port builds but never reads back (backlog#1823)"
)]
pub struct Initiator {
id: String,
display_name: String,
}
pub struct CopyObjectResult {
pub etag: String,
pub last_modified: OffsetDateTime,
}
#[derive(Debug, Clone)]
pub struct ObjectPart {
pub etag: String,
@@ -263,7 +260,6 @@ pub struct CompletePart {
}
impl CompletePart {
#[allow(dead_code, reason = "MinIO-parity accessor with no caller in this port (backlog#1823)")]
fn checksum(&self, t: &ChecksumMode) -> String {
match t {
ChecksumMode::ChecksumCRC32C => {
@@ -288,6 +284,11 @@ impl CompletePart {
}
}
pub struct CopyObjectPartResult {
pub etag: String,
pub last_modified: OffsetDateTime,
}
#[derive(Debug, Default, serde::Serialize)]
#[serde(rename = "CompleteMultipartUpload")]
pub struct CompleteMultipartUpload {
@@ -356,10 +357,10 @@ impl CompleteMultipartUpload {
}
}
#[allow(
dead_code,
reason = "live via quick_xml::de::from_str in bucket_cache.rs; serde deserialization is not a construction (backlog#1823)"
)]
pub struct CreateBucketConfiguration {
pub location: String,
}
#[derive(serde::Serialize)]
pub struct DeleteObject {
//api has
@@ -367,6 +368,21 @@ pub struct DeleteObject {
pub version_id: String,
}
pub struct DeletedObject {
//s3s has
pub key: String,
pub version_id: String,
pub deletemarker: bool,
pub deletemarker_version_id: String,
}
pub struct NonDeletedObject {
pub key: String,
pub code: String,
pub message: String,
pub version_id: String,
}
#[derive(serde::Serialize)]
pub struct DeleteMultiObjects {
pub quiet: bool,
@@ -386,7 +402,6 @@ impl DeleteMultiObjects {
Ok(buf)
}
#[allow(dead_code, reason = "MinIO-parity XML helper with no caller in this port (backlog#1823)")]
pub fn unmarshal(buf: &[u8]) -> Result<Self, std::io::Error> {
#[derive(Debug, Deserialize)]
struct WireDeleteObject {
@@ -421,3 +436,8 @@ impl DeleteMultiObjects {
})
}
}
pub struct DeleteMultiObjectsResult {
pub deleted_objects: Vec<DeletedObject>,
pub undeleted_objects: Vec<NonDeletedObject>,
}
+3 -7
View File
@@ -365,15 +365,11 @@ mod tests {
pub struct Checksum {
checksum_type: ChecksumMode,
r: Vec<u8>,
#[allow(
dead_code,
reason = "checksum bookkeeping field kept beside the value it guards (backlog#1823)"
)]
computed: bool,
}
#[allow(dead_code)]
impl Checksum {
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
fn new(t: ChecksumMode, b: &[u8]) -> Checksum {
if t.is_set() && b.len() == t.raw_byte_len() {
return Checksum {
@@ -385,7 +381,7 @@ impl Checksum {
Checksum::default()
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
#[allow(dead_code)]
fn new_checksum_string(t: ChecksumMode, s: &str) -> Result<Checksum, std::io::Error> {
let b = match base64_decode(s.as_bytes()) {
Ok(b) => b,
@@ -412,7 +408,7 @@ impl Checksum {
base64_encode(&self.r)
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
#[allow(dead_code)]
fn raw(&self) -> Option<Vec<u8>> {
if !self.is_set() {
return None;
+3
View File
@@ -32,5 +32,8 @@ pub const MAX_MULTIPART_PUT_OBJECT_SIZE: i64 = 1024 * 1024 * 1024 * 1024 * 5;
pub const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD";
pub const UNSIGNED_PAYLOAD_TRAILER: &str = "STREAMING-UNSIGNED-PAYLOAD-TRAILER";
pub const TOTAL_WORKERS: i64 = 4;
pub const SIGN_V4_ALGORITHM: &str = "AWS4-HMAC-SHA256";
pub const ISO8601_DATEFORMAT: &[FormatItem<'_>] =
format_description!("[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond]Z");
+19 -12
View File
@@ -67,10 +67,6 @@ impl<P: Provider + Default> Credentials<P> {
Ok(self.creds.clone())
}
#[allow(
dead_code,
reason = "MinIO-parity credential surface with no caller in this port (backlog#1823)"
)]
fn expire(&mut self) {
self.force_refresh = true;
}
@@ -137,10 +133,6 @@ impl Provider for Static {
#[derive(Debug, Clone, Default)]
pub struct STSError {
#[allow(
dead_code,
reason = "MinIO-parity STS error detail that this port never reads back (backlog#1823)"
)]
pub r#type: String,
pub code: String,
pub message: String,
@@ -149,10 +141,6 @@ pub struct STSError {
#[derive(Debug, Clone, thiserror::Error)]
pub struct ErrorResponse {
pub sts_error: STSError,
#[allow(
dead_code,
reason = "MinIO-parity STS error detail that this port never reads back (backlog#1823)"
)]
pub request_id: String,
}
@@ -170,3 +158,22 @@ impl ErrorResponse {
return self.sts_error.message.clone();
}
}
pub fn xml_decoder<T>(body: &[u8]) -> Result<T, Error>
where
for<'de> T: Deserialize<'de>,
{
match std::str::from_utf8(body) {
Ok(xml_body) => quick_xml::de::from_str::<T>(xml_body).map_err(|err| Error::new(ErrorKind::InvalidData, err.to_string())),
Err(err) => Err(Error::new(ErrorKind::InvalidData, err.to_string())),
}
}
pub fn xml_decode_and_body<T>(body_reader: &[u8]) -> Result<(Vec<u8>, T), std::io::Error>
where
for<'de> T: Deserialize<'de>,
{
let body = body_reader.to_vec();
let parsed = xml_decoder(&body)?;
Ok((body, parsed))
}
+1
View File
@@ -13,6 +13,7 @@
// limitations under the License.
// #730: S3 client compatibility models are kept while ECStore callers move to narrower facades.
#![allow(dead_code)]
pub mod admin_handler_utils;
pub mod api_error_response;
+34 -2
View File
@@ -37,17 +37,16 @@ pub struct PutObjReader {
//pub sealMD5Fn: SealMD5CurrFn,
}
#[allow(dead_code)]
impl PutObjReader {
pub fn new(reader: HashReader) -> Self {
Self { reader }
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
fn md5_current_hex_string(&self) -> String {
self.reader.checksum().map(|v| v.encoded).unwrap_or_default()
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
fn with_encryption(&mut self, enc_reader: HashReader) -> Result<(), std::io::Error> {
self.reader = enc_reader;
@@ -78,6 +77,39 @@ fn part_number_to_rangespec(oi: ObjectInfo, part_number: usize) -> Option<HTTPRa
})
}
fn get_compressed_offsets(oi: ObjectInfo, offset: i64) -> (i64, i64, i64, i64, u64) {
let mut skip_length: i64 = 0;
let mut cumulative_actual_size: i64 = 0;
let mut first_part_idx: i64 = 0;
let mut compressed_offset: i64 = 0;
let mut part_skip: i64 = 0;
let mut decrypt_skip: i64 = 0;
let mut seq_num: u64 = 0;
for (i, part) in oi.parts.iter().enumerate() {
cumulative_actual_size += part.actual_size as i64;
if cumulative_actual_size <= offset {
compressed_offset += part.size as i64;
} else {
first_part_idx = i as i64;
skip_length = cumulative_actual_size - part.actual_size as i64;
break;
}
}
skip_length = offset - skip_length;
let parts: &[ObjectPartInfo] = &oi.parts;
if skip_length > 0
&& parts.len() > first_part_idx as usize
&& parts[first_part_idx as usize].index.as_ref().is_some_and(|idx| idx.len() > 0)
{
let _ = part_skip;
let _ = decrypt_skip;
let _ = seq_num;
}
(compressed_offset, part_skip, first_part_idx, decrypt_skip, seq_num)
}
pub fn new_getobjectreader<'a>(
rs: &Option<HTTPRangeSpec>,
oi: &'a ObjectInfo,
@@ -23,7 +23,6 @@ const X_OBS_VERSION_ID: &str = "x-obs-version-id";
const MAX_REMOTE_VERSION_ID_LEN: usize = 1024;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[allow(dead_code, reason = "bucket versioning states kept as a complete vocabulary (backlog#1823)")]
pub(crate) enum BucketVersioningState {
Unknown,
Disabled,
@@ -48,7 +47,6 @@ impl RemoteVersion {
}
}
#[allow(dead_code, reason = "MinIO-parity accessor with no caller in this port (backlog#1823)")]
pub(crate) fn exact_request_id(&self) -> Result<Option<&str>, Error> {
match self {
Self::Unknown => Err(Error::new(
@@ -101,10 +101,6 @@ where
const C_UNKNOWN: i32 = -1;
const C_OFFLINE: i32 = 0;
#[allow(
dead_code,
reason = "reachable only from the unused transition client methods below (backlog#1823)"
)]
const C_ONLINE: i32 = 1;
fn invalid_utf8_header_error(scope: &str, header_name: &str) -> std::io::Error {
@@ -324,10 +320,6 @@ impl TransitionClient {
Ok(client)
}
#[allow(
dead_code,
reason = "MinIO-parity transition client surface with no caller in this port (backlog#1823)"
)]
fn endpoint_url(&self) -> Url {
self.endpoint_url.clone()
}
@@ -356,20 +348,12 @@ impl TransitionClient {
.to_string())
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn trace_errors_only_off(&self) {
if let Ok(mut trace_errors_only) = self.trace_errors_only.lock() {
*trace_errors_only = false;
}
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn trace_off(&self) {
if let Ok(mut is_trace_enabled) = self.is_trace_enabled.lock() {
*is_trace_enabled = false;
@@ -379,20 +363,12 @@ impl TransitionClient {
}
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn set_s3_transfer_accelerate(&self, accelerate_endpoint: &str) {
if let Ok(mut endpoint) = self.s3_accelerate_endpoint.lock() {
*endpoint = accelerate_endpoint.to_string();
}
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn set_s3_enable_dual_stack(&self, enabled: bool) {
if let Ok(mut dual_stack) = self.s3_dual_stack_enabled.lock() {
*dual_stack = enabled;
@@ -422,18 +398,10 @@ impl TransitionClient {
(hash_algos, hash_sums)
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn is_online(&self) -> bool {
!self.is_offline()
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn mark_offline(&self) {
self.health_status
.compare_exchange(C_ONLINE, C_OFFLINE, Ordering::SeqCst, Ordering::SeqCst);
@@ -443,18 +411,10 @@ impl TransitionClient {
self.health_status.load(Ordering::SeqCst) == C_OFFLINE
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn health_check(hc_duration: Duration) {
let _ = hc_duration;
}
#[allow(
dead_code,
reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
)]
fn dump_http(&self, req: &Request<s3s::Body>, resp: &Response<Incoming>) -> Result<(), std::io::Error> {
let mut resp_trace: Vec<u8>;
@@ -1142,7 +1102,6 @@ impl Default for ObjectInfo {
}
impl ObjectInfo {
#[allow(dead_code, reason = "MinIO-parity accessor with no caller in this port (backlog#1823)")]
pub(crate) fn remote_version(
&self,
capabilities: ProviderVersionCapabilities,
+4
View File
@@ -48,6 +48,10 @@ lazy_static! {
};
}
pub fn is_standard_query_value(qs_key: &str) -> bool {
SUPPORTED_QUERY_VALUES[qs_key]
}
pub fn is_storageclass_header(header_key: &str) -> bool {
header_key.to_lowercase() == X_AMZ_STORAGE_CLASS.as_str().to_lowercase()
}
+461 -22
View File
@@ -59,7 +59,7 @@ use std::{
path::PathBuf,
sync::{
Arc,
atomic::{AtomicU32, Ordering},
atomic::{AtomicBool, AtomicU32, Ordering},
},
time::Duration,
};
@@ -216,6 +216,8 @@ where
#[derive(Debug)]
pub struct RemoteDisk {
/// Stable identity for this handle instance; replacement handles receive a new identity.
handle_id: Uuid,
pub id: Mutex<Option<Uuid>>,
pub addr: String,
endpoint: Endpoint,
@@ -226,9 +228,22 @@ pub struct RemoteDisk {
health: Arc<DiskHealthTracker>,
/// Cancellation token for monitoring tasks
cancel_token: CancellationToken,
recovery_monitor_active: Arc<AtomicBool>,
#[cfg(test)]
recovery_monitor_start_count: Arc<AtomicU32>,
data_transport: Arc<dyn InternodeDataTransport>,
}
struct RecoveryMonitorLease {
active: Arc<AtomicBool>,
}
impl Drop for RecoveryMonitorLease {
fn drop(&mut self) {
self.active.store(false, Ordering::Release);
}
}
// ── Connection lifecycle (grpc-optimization P3) ──
/// Whether to prewarm the internode control channel in the background at construction (default off).
@@ -368,14 +383,15 @@ impl RemoteDisk {
.await
}
fn recovery_monitor_span(addr: &str, endpoint: &Endpoint) -> tracing::Span {
fn recovery_monitor_span(addr: &str, endpoint: &Endpoint, handle_id: Uuid) -> tracing::Span {
tracing::info_span!(
"recovery-monitor",
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
kind = "remote_disk",
endpoint = %endpoint,
addr = %addr
addr = %addr,
handle_id = %handle_id
)
}
@@ -411,6 +427,7 @@ impl RemoteDisk {
rustfs_utils::get_env_bool(ENV_RUSTFS_DRIVE_ACTIVE_MONITORING, DEFAULT_RUSTFS_DRIVE_ACTIVE_MONITORING);
let disk = Self {
handle_id: Uuid::new_v4(),
id: Mutex::new(None),
addr,
endpoint: ep.clone(),
@@ -418,6 +435,9 @@ impl RemoteDisk {
health_check: opt.health_check && env_health_check,
health: Arc::new(DiskHealthTracker::new()),
cancel_token: CancellationToken::new(),
recovery_monitor_active: Arc::new(AtomicBool::new(false)),
#[cfg(test)]
recovery_monitor_start_count: Arc::new(AtomicU32::new(0)),
data_transport,
};
record_drive_runtime_state(ep, RuntimeDriveHealthState::Online);
@@ -435,6 +455,16 @@ impl RemoteDisk {
self.health.runtime_state()
}
#[cfg(test)]
fn recovery_monitor_is_active(&self) -> bool {
self.recovery_monitor_active.load(Ordering::Acquire)
}
#[cfg(test)]
fn recovery_monitor_start_count(&self) -> u32 {
self.recovery_monitor_start_count.load(Ordering::Acquire)
}
pub fn offline_duration_secs(&self) -> Option<u64> {
self.health.offline_duration().map(|duration| duration.as_secs())
}
@@ -573,13 +603,54 @@ impl RemoteDisk {
return;
}
let addr = self.addr.clone();
let endpoint = self.endpoint.clone();
let health = Arc::clone(&self.health);
let cancel_token = self.cancel_token.clone();
let span = Self::recovery_monitor_span(&addr, &endpoint);
Self::schedule_recovery_monitor(
self.addr.clone(),
self.endpoint.clone(),
self.handle_id,
Arc::clone(&self.health),
self.cancel_token.clone(),
Arc::clone(&self.recovery_monitor_active),
#[cfg(test)]
Arc::clone(&self.recovery_monitor_start_count),
);
}
fn schedule_recovery_monitor(
addr: String,
endpoint: Endpoint,
handle_id: Uuid,
health: Arc<DiskHealthTracker>,
cancel_token: CancellationToken,
active: Arc<AtomicBool>,
#[cfg(test)] start_count: Arc<AtomicU32>,
) {
if active
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_err()
{
return;
}
let span = Self::recovery_monitor_span(&addr, &endpoint, handle_id);
super::spawn_background_monitor(span, async move {
Self::monitor_remote_disk_recovery(addr, endpoint, health, cancel_token).await;
#[cfg(test)]
start_count.fetch_add(1, Ordering::AcqRel);
let lease = RecoveryMonitorLease {
active: Arc::clone(&active),
};
Self::monitor_remote_disk_recovery(addr.clone(), endpoint.clone(), Arc::clone(&health), cancel_token.clone()).await;
drop(lease);
if !cancel_token.is_cancelled() && health.runtime_state() != RuntimeDriveHealthState::Online {
Self::schedule_recovery_monitor(
addr,
endpoint,
handle_id,
health,
cancel_token,
active,
#[cfg(test)]
start_count,
);
}
});
}
@@ -588,7 +659,7 @@ impl RemoteDisk {
let (tx, rx) = tokio::sync::oneshot::channel();
let endpoint = self.endpoint.clone();
let addr = self.addr.clone();
let span = Self::recovery_monitor_span(&addr, &endpoint);
let span = Self::recovery_monitor_span(&addr, &endpoint, self.handle_id);
super::spawn_background_monitor(span, async move {
warn!(
event = EVENT_REMOTE_DISK_HEALTH,
@@ -619,9 +690,11 @@ impl RemoteDisk {
let cancel_token = self.cancel_token.clone();
let addr = self.addr.clone();
let endpoint = self.endpoint.clone();
let handle_id = self.handle_id;
let recovery_monitor_active = Arc::clone(&self.recovery_monitor_active);
tokio::spawn(async move {
Self::monitor_remote_disk_health(addr, endpoint, health, cancel_token).await;
Self::monitor_remote_disk_health(addr, endpoint, handle_id, health, cancel_token, recovery_monitor_active).await;
});
}
@@ -629,8 +702,10 @@ impl RemoteDisk {
async fn monitor_remote_disk_health(
addr: String,
endpoint: Endpoint,
handle_id: Uuid,
health: Arc<DiskHealthTracker>,
cancel_token: CancellationToken,
recovery_monitor_active: Arc<AtomicBool>,
) {
let mut interval = time::interval(get_drive_active_check_interval());
@@ -655,11 +730,16 @@ impl RemoteDisk {
let addr_clone = addr.clone();
let endpoint_clone = endpoint.clone();
let cancel_clone = cancel_token.clone();
let span = Self::recovery_monitor_span(&addr_clone, &endpoint_clone);
super::spawn_background_monitor(span, async move {
Self::monitor_remote_disk_recovery(addr_clone, endpoint_clone, health_clone, cancel_clone).await;
});
Self::schedule_recovery_monitor(
addr_clone,
endpoint_clone,
handle_id,
health_clone,
cancel_clone,
Arc::clone(&recovery_monitor_active),
#[cfg(test)]
Arc::new(AtomicU32::new(0)),
);
}
loop {
@@ -718,11 +798,16 @@ impl RemoteDisk {
let addr_clone = addr.clone();
let endpoint_clone = endpoint.clone();
let cancel_clone = cancel_token.clone();
let span = Self::recovery_monitor_span(&addr_clone, &endpoint_clone);
super::spawn_background_monitor(span, async move {
Self::monitor_remote_disk_recovery(addr_clone, endpoint_clone, health_clone, cancel_clone).await;
});
Self::schedule_recovery_monitor(
addr_clone,
endpoint_clone,
handle_id,
health_clone,
cancel_clone,
Arc::clone(&recovery_monitor_active),
#[cfg(test)]
Arc::new(AtomicU32::new(0)),
);
}
}
}
@@ -973,6 +1058,7 @@ impl RemoteDisk {
subsystem = LOG_SUBSYSTEM_REMOTE_DISK,
endpoint = %self.endpoint,
addr = %self.addr,
handle_id = %self.handle_id,
op,
state = "faulty_short_circuit",
"Remote disk operation short-circuited by faulty state"
@@ -3116,15 +3202,23 @@ mod tests {
use super::*;
use crate::cluster::rpc::internode_data_transport::{InternodeDataTransportCapabilities, TcpHttpInternodeDataTransport};
use crate::runtime::sources as runtime_sources;
use rustfs_protos::proto_gen::node_service::{DiskInfoResponse, ReadAllResponse};
use serde_json::Value;
use serial_test::serial;
use std::convert::Infallible;
use std::future::Future;
use std::io::{self as std_io, Write};
use std::pin::Pin;
use std::sync::{Arc, Mutex, Mutex as StdMutex, Once};
use std::task::{Context, Poll};
use tokio::io::{ReadBuf, duplex};
use tokio::net::TcpListener;
use tonic::transport::Endpoint as TonicEndpoint;
use tonic::transport::{Endpoint as TonicEndpoint, Server};
use tonic::{Response, Status};
use tonic::{
codegen::{Body as HttpBody, BoxFuture, StdError, http},
server::NamedService,
};
use tracing::Level;
use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt};
use uuid::Uuid;
@@ -3284,6 +3378,205 @@ mod tests {
ns_scanner_probe_status: Arc<StdMutex<Option<u16>>>,
}
#[derive(Clone, Debug)]
struct AuthenticatedReadPeer {
audience: String,
disk_info_calls: Arc<AtomicU32>,
read_all_calls: Arc<AtomicU32>,
read_all_disks: Arc<StdMutex<Vec<String>>>,
read_all_data: Bytes,
}
impl AuthenticatedReadPeer {
fn new(audience: String, read_all_data: Bytes) -> Self {
Self {
audience,
disk_info_calls: Arc::new(AtomicU32::new(0)),
read_all_calls: Arc::new(AtomicU32::new(0)),
read_all_disks: Arc::default(),
read_all_data,
}
}
fn disk_info_calls(&self) -> u32 {
self.disk_info_calls.load(Ordering::Acquire)
}
fn read_all_calls(&self) -> u32 {
self.read_all_calls.load(Ordering::Acquire)
}
fn read_all_disks(&self) -> Vec<String> {
self.read_all_disks.lock().expect("read_all disk list lock poisoned").clone()
}
fn verify_auth<T>(&self, request: &Request<T>, path: &str) -> std::result::Result<(), Status> {
let headers = request.metadata().clone().into_headers();
crate::cluster::rpc::verify_tonic_rpc_signature(&self.audience, path, &headers)
.map_err(|err| Status::unauthenticated(err.to_string()))
}
}
#[derive(Clone, Debug)]
struct AuthenticatedReadPeerService {
peer: AuthenticatedReadPeer,
}
impl NamedService for AuthenticatedReadPeerService {
const NAME: &'static str = "node_service.NodeService";
}
impl<B> tower::Service<http::Request<B>> for AuthenticatedReadPeerService
where
B: HttpBody + Send + 'static,
B::Error: Into<StdError> + Send + 'static,
{
type Response = http::Response<tonic::body::Body>;
type Error = Infallible;
type Future = BoxFuture<Self::Response, Self::Error>;
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, request: http::Request<B>) -> Self::Future {
match request.uri().path() {
"/node_service.NodeService/DiskInfo" => {
#[derive(Clone)]
struct DiskInfoSvc(AuthenticatedReadPeer);
impl tonic::server::UnaryService<DiskInfoRequest> for DiskInfoSvc {
type Response = DiskInfoResponse;
type Future = Pin<Box<dyn Future<Output = std::result::Result<Response<Self::Response>, Status>> + Send>>;
fn call(&mut self, request: Request<DiskInfoRequest>) -> Self::Future {
let peer = self.0.clone();
Box::pin(async move {
peer.verify_auth(&request, "/node_service.NodeService/DiskInfo")?;
let request = request.into_inner();
let opts = serde_json::from_str::<DiskInfoOptions>(&request.opts)
.map_err(|err| Status::invalid_argument(err.to_string()))?;
if !opts.noop {
return Err(Status::invalid_argument("recovery probe must use noop disk_info"));
}
peer.disk_info_calls.fetch_add(1, Ordering::AcqRel);
let disk_info = serde_json::to_string(&DiskInfo {
total: 1,
free: 1,
endpoint: request.disk,
..Default::default()
})
.map_err(|err| Status::internal(err.to_string()))?;
Ok(Response::new(DiskInfoResponse {
success: true,
disk_info,
error: None,
}))
})
}
}
let peer = self.peer.clone();
Box::pin(async move {
let method = DiskInfoSvc(peer);
let codec = tonic_prost::ProstCodec::default();
let mut grpc = tonic::server::Grpc::new(codec);
Ok(grpc.unary(method, request).await)
})
}
"/node_service.NodeService/ReadAll" => {
#[derive(Clone)]
struct ReadAllSvc(AuthenticatedReadPeer);
impl tonic::server::UnaryService<ReadAllRequest> for ReadAllSvc {
type Response = ReadAllResponse;
type Future = Pin<Box<dyn Future<Output = std::result::Result<Response<Self::Response>, Status>> + Send>>;
fn call(&mut self, request: Request<ReadAllRequest>) -> Self::Future {
let peer = self.0.clone();
Box::pin(async move {
peer.verify_auth(&request, "/node_service.NodeService/ReadAll")?;
let request = request.into_inner();
peer.read_all_calls.fetch_add(1, Ordering::AcqRel);
peer.read_all_disks
.lock()
.expect("read_all disk list lock poisoned")
.push(request.disk);
Ok(Response::new(ReadAllResponse {
success: true,
data: peer.read_all_data.clone(),
error: None,
}))
})
}
}
let peer = self.peer.clone();
Box::pin(async move {
let method = ReadAllSvc(peer);
let codec = tonic_prost::ProstCodec::default();
let mut grpc = tonic::server::Grpc::new(codec);
Ok(grpc.unary(method, request).await)
})
}
_ => Box::pin(async move {
let mut response = http::Response::new(tonic::body::Body::default());
let headers = response.headers_mut();
headers.insert(tonic::Status::GRPC_STATUS, (tonic::Code::Unimplemented as i32).into());
headers.insert(http::header::CONTENT_TYPE, tonic::metadata::GRPC_CONTENT_TYPE);
Ok(response)
}),
}
}
}
struct TestGrpcPeer {
addr: String,
peer: AuthenticatedReadPeer,
shutdown: CancellationToken,
task: tokio::task::JoinHandle<()>,
}
impl TestGrpcPeer {
async fn spawn(read_all_data: Bytes) -> Option<Self> {
let listener = match TcpListener::bind("127.0.0.1:0").await {
Ok(listener) => listener,
Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None,
Err(err) => panic!("test gRPC listener should bind: {err}"),
};
let socket_addr = listener.local_addr().expect("listener local address should be available");
let addr = format!("http://{socket_addr}");
let audience = crate::cluster::rpc::normalize_tonic_rpc_audience(&socket_addr.to_string())
.expect("test audience should normalize");
let peer = AuthenticatedReadPeer::new(audience, read_all_data);
let service = AuthenticatedReadPeerService { peer: peer.clone() };
let shutdown = CancellationToken::new();
let shutdown_for_task = shutdown.clone();
let incoming = futures_util::stream::unfold(listener, |listener| async {
Some((listener.accept().await.map(|(stream, _)| stream), listener))
});
let task = tokio::spawn(async move {
Server::builder()
.add_service(service)
.serve_with_incoming_shutdown(incoming, shutdown_for_task.cancelled_owned())
.await
.expect("test gRPC peer should serve");
});
Some(Self {
addr,
peer,
shutdown,
task,
})
}
async fn stop(self) {
self.shutdown.cancel();
let _ = self.task.await;
}
}
impl RecordingInternodeDataTransport {
fn with_ns_scanner_probe_status(status: u16) -> Self {
Self {
@@ -4397,6 +4690,152 @@ mod tests {
accept_task.abort();
}
#[tokio::test]
async fn faulty_handle_runs_only_one_recovery_monitor() {
let endpoint = Endpoint {
url: url::Url::parse("http://remote-node:9000/data/rustfs0").expect("endpoint should parse"),
is_local: false,
pool_idx: 0,
set_idx: 0,
disk_idx: 0,
};
let disk = RemoteDisk::new(
&endpoint,
&DiskOption {
cleanup: false,
health_check: true,
},
Arc::new(TcpHttpInternodeDataTransport),
)
.await
.expect("remote disk should construct");
if !disk.health_check {
return;
}
disk.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
disk.spawn_recovery_monitor_if_needed();
disk.spawn_recovery_monitor_if_needed();
tokio::time::timeout(Duration::from_secs(1), async {
while disk.recovery_monitor_start_count() == 0 {
tokio::task::yield_now().await;
}
})
.await
.expect("recovery monitor should start");
assert!(disk.recovery_monitor_is_active(), "only one recovery monitor should own the handle");
assert_eq!(
disk.recovery_monitor_start_count(),
1,
"the failed compare-exchange path must not start a second monitor"
);
disk.cancel_token.cancel();
tokio::time::timeout(Duration::from_secs(1), async {
while disk.recovery_monitor_is_active() {
tokio::task::yield_now().await;
}
})
.await
.expect("cancelled recovery monitor should release its single-flight state");
assert!(!disk.recovery_monitor_is_active());
}
#[tokio::test]
#[serial(remote_disk_recovery_probe)]
async fn recovery_monitor_restores_online_then_real_reads_use_replacement_handle() {
runtime_sources::ensure_test_rpc_secret();
let Some(peer) = TestGrpcPeer::spawn(Bytes::from_static(b"replacement-data")).await else {
return;
};
let url = url::Url::parse(&format!("{}/data/rustfs0", peer.addr)).expect("endpoint should parse");
let endpoint = Endpoint {
url,
is_local: false,
pool_idx: 0,
set_idx: 0,
disk_idx: 0,
};
let disk = RemoteDisk::new(
&endpoint,
&DiskOption {
cleanup: false,
health_check: true,
},
Arc::new(TcpHttpInternodeDataTransport),
)
.await
.expect("remote disk should construct");
disk.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
temp_env::async_with_vars(
[
(rustfs_config::ENV_DRIVE_RETURNING_PROBE_INTERVAL_SECS, Some("1")),
(rustfs_config::ENV_DRIVE_RETURNING_SUCCESS_THRESHOLD, Some("3")),
(rustfs_config::ENV_DRIVE_ACTIVE_CHECK_TIMEOUT_SECS, Some("1")),
],
async {
let monitor = tokio::spawn(RemoteDisk::monitor_remote_disk_recovery(
disk.addr.clone(),
endpoint.clone(),
Arc::clone(&disk.health),
disk.cancel_token.clone(),
));
tokio::time::timeout(Duration::from_secs(5), async {
while disk.runtime_state() != RuntimeDriveHealthState::Online {
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await
.expect("three authenticated recovery probes should restore the disk online");
monitor.await.expect("recovery monitor should exit after restoring Online");
assert_eq!(
peer.peer.disk_info_calls(),
3,
"RemoteDisk recovery requires the configured three successful disk_info probes"
);
let recovered_read = disk.read_all("bucket", "object").await.expect("recovered handle should read");
assert_eq!(recovered_read, Bytes::from_static(b"replacement-data"));
let replacement = RemoteDisk::new(
&endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
Arc::new(TcpHttpInternodeDataTransport),
)
.await
.expect("replacement remote disk should construct");
let replacement_id = Uuid::new_v4();
replacement
.set_disk_id(Some(replacement_id))
.await
.expect("replacement disk id should set");
let replacement_read = replacement
.read_all("bucket", "object")
.await
.expect("replacement handle should route real reads");
assert_eq!(replacement_read, Bytes::from_static(b"replacement-data"));
assert_eq!(peer.peer.read_all_calls(), 2);
assert_eq!(
peer.peer.read_all_disks(),
vec![endpoint.to_string(), replacement_id.to_string()],
"real reads must use the current handle's disk reference"
);
disk.cancel_token.cancel();
replacement.cancel_token.cancel();
},
)
.await;
peer.stop().await;
}
#[tokio::test]
async fn test_copy_stream_with_buffer_copies_full_payload() {
let payload = b"walk-dir-stream".repeat(1024);
+3
View File
@@ -39,6 +39,7 @@ use rustfs_config::{
};
use std::sync::LazyLock;
#[allow(dead_code)]
#[allow(clippy::declare_interior_mutable_const)]
/// Default KVS for audit webhook settings.
pub static DEFAULT_AUDIT_WEBHOOK_KVS: LazyLock<KVS> = LazyLock::new(|| {
@@ -116,6 +117,7 @@ pub static DEFAULT_AUDIT_WEBHOOK_KVS: LazyLock<KVS> = LazyLock::new(|| {
])
});
#[allow(dead_code)]
#[allow(clippy::declare_interior_mutable_const)]
/// Default KVS for audit MQTT settings.
pub static DEFAULT_AUDIT_MQTT_KVS: LazyLock<KVS> = LazyLock::new(|| {
@@ -373,6 +375,7 @@ pub static DEFAULT_AUDIT_NATS_KVS: LazyLock<KVS> = LazyLock::new(|| {
])
});
#[allow(dead_code)]
pub static DEFAULT_AUDIT_PULSAR_KVS: LazyLock<KVS> = LazyLock::new(|| {
KVS(vec![
KV {
+59
View File
@@ -12,9 +12,12 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::error::{Error, Result};
use rustfs_config::server_config::{KV, KVS};
use rustfs_config::{DEFAULT_HEAL_BITROT_CYCLE_SECS, HEAL_BITROT_CYCLE};
use rustfs_utils::string::parse_bool;
use std::sync::LazyLock;
use std::time::Duration;
pub static DEFAULT_KVS: LazyLock<KVS> = LazyLock::new(|| {
KVS(vec![KV {
@@ -23,3 +26,59 @@ pub static DEFAULT_KVS: LazyLock<KVS> = LazyLock::new(|| {
hidden_if_empty: false,
}])
});
#[derive(Debug, Default)]
pub struct Config {
pub bitrot: String,
pub sleep: Duration,
pub io_count: usize,
pub drive_workers: usize,
pub cache: Duration,
}
impl Config {
pub fn bitrot_scan_cycle(&self) -> Duration {
self.cache
}
pub fn get_workers(&self) -> usize {
self.drive_workers
}
pub fn update(&mut self, nopts: &Config) {
self.bitrot = nopts.bitrot.clone();
self.io_count = nopts.io_count;
self.sleep = nopts.sleep;
self.drive_workers = nopts.drive_workers;
}
}
const RUSTFS_BITROT_CYCLE_IN_MONTHS: u64 = 1;
fn parse_bitrot_config(s: &str) -> Result<Duration> {
match parse_bool(s) {
Ok(enabled) => {
if enabled {
Ok(Duration::from_secs_f64(0.0))
} else {
Ok(Duration::from_secs_f64(-1.0))
}
}
Err(_) => {
if !s.ends_with("m") {
return Err(Error::other("unknown format"));
}
match s.trim_end_matches('m').parse::<u64>() {
Ok(months) => {
if months < RUSTFS_BITROT_CYCLE_IN_MONTHS {
return Err(Error::other(format!("minimum bitrot cycle is {RUSTFS_BITROT_CYCLE_IN_MONTHS} month(s)")));
}
Ok(Duration::from_secs(months * 30 * 24 * 60))
}
Err(err) => Err(Error::other(err)),
}
}
}
}
+1
View File
@@ -16,6 +16,7 @@
mod audit;
pub mod com;
#[allow(dead_code)]
pub mod heal;
mod notify;
mod oidc;
+3 -9
View File
@@ -1996,11 +1996,11 @@ impl PoolMeta {
Ok(false)
}
#[allow(dead_code)]
pub fn validate(&self, pools: Vec<Arc<Sets>>) -> Result<bool> {
struct PoolInfo {
position: usize,
completed: bool,
#[allow(dead_code, reason = "written but never read back (backlog#1823)")]
decom_started: bool,
}
@@ -4958,19 +4958,13 @@ fn is_disk_online_state(state: &str) -> bool {
}
#[deprecated(since = "0.1.0", note = "Use fallback_total_capacity_dedup instead")]
#[allow(
dead_code,
reason = "superseded by the replacement named in the comment at pools.rs:5071 (backlog#1823)"
)]
#[allow(dead_code)]
fn fallback_total_capacity(disks: &[rustfs_madmin::Disk]) -> usize {
fallback_total_capacity_dedup(disks)
}
#[deprecated(since = "0.1.0", note = "Use fallback_free_capacity_dedup instead")]
#[allow(
dead_code,
reason = "superseded by the replacement named in the comment at pools.rs:5071 (backlog#1823)"
)]
#[allow(dead_code)]
fn fallback_free_capacity(disks: &[rustfs_madmin::Disk]) -> usize {
fallback_free_capacity_dedup(disks)
}
@@ -12,16 +12,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//! Per-disk usage snapshots persisted under the metadata bucket.
//!
//! **Nothing calls into this module.** It landed complete with tests in #5307
//! (2026-07-27) and its aggregation entry point,
//! [`crate::data_usage::aggregate_local_snapshots`], has never had a caller in
//! the tree's history. The live data-usage path is
//! `load_data_usage_from_backend` / `store_data_usage_in_backend`. The items
//! below therefore carry individual `dead_code` allows rather than a module
//! blanket, so the gap stays greppable until it is either wired up or removed.
use crate::data_usage::BucketUsageInfo;
use crate::disk::RUSTFS_META_BUCKET;
use crate::error::{Error, Result};
@@ -36,12 +26,10 @@ pub const DATA_USAGE_DIR: &str = "datausage";
/// Directory used to store incremental scan state files under the metadata bucket.
pub const DATA_USAGE_STATE_DIR: &str = "datausage/state";
/// Snapshot file format version, allows forward compatibility if the structure evolves.
#[allow(dead_code, reason = "unwired local usage-snapshot feature; see module docs (backlog#1823)")]
pub const LOCAL_USAGE_SNAPSHOT_VERSION: u32 = 1;
/// Additional metadata describing which disk produced the snapshot.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
#[allow(dead_code, reason = "unwired local usage-snapshot feature; see module docs (backlog#1823)")]
pub struct LocalUsageSnapshotMeta {
/// Disk UUID stored as a string for simpler serialization.
pub disk_id: String,
@@ -55,7 +43,6 @@ pub struct LocalUsageSnapshotMeta {
/// Usage snapshot produced by a single disk.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
#[allow(dead_code, reason = "unwired local usage-snapshot feature; see module docs (backlog#1823)")]
pub struct LocalUsageSnapshot {
/// Format version recorded in the snapshot.
pub format_version: u32,
@@ -77,7 +64,6 @@ pub struct LocalUsageSnapshot {
pub objects_total_size: u64,
}
#[allow(dead_code, reason = "unwired local usage-snapshot feature; see module docs (backlog#1823)")]
impl LocalUsageSnapshot {
/// Create an empty snapshot with the default format version filled in.
pub fn new(meta: LocalUsageSnapshotMeta) -> Self {
@@ -113,13 +99,11 @@ impl LocalUsageSnapshot {
}
/// Build the snapshot file name `<disk-id>.json`.
#[allow(dead_code, reason = "unwired local usage-snapshot feature; see module docs (backlog#1823)")]
pub fn snapshot_file_name(disk_id: &str) -> String {
format!("{disk_id}.json")
}
/// Build the object path relative to `RUSTFS_META_BUCKET`, e.g. `datausage/<disk-id>.json`.
#[allow(dead_code, reason = "unwired local usage-snapshot feature; see module docs (backlog#1823)")]
pub fn snapshot_object_path(disk_id: &str) -> String {
format!("{}/{}", DATA_USAGE_DIR, snapshot_file_name(disk_id))
}
@@ -135,13 +119,11 @@ pub fn data_usage_state_dir(root: &Path) -> PathBuf {
}
/// Build the absolute path to the snapshot file for the provided disk ID.
#[allow(dead_code, reason = "unwired local usage-snapshot feature; see module docs (backlog#1823)")]
pub fn snapshot_path(root: &Path, disk_id: &str) -> PathBuf {
data_usage_dir(root).join(snapshot_file_name(disk_id))
}
/// Read a snapshot from disk if it exists.
#[allow(dead_code, reason = "unwired local usage-snapshot feature; see module docs (backlog#1823)")]
pub async fn read_snapshot(root: &Path, disk_id: &str) -> Result<Option<LocalUsageSnapshot>> {
let path = snapshot_path(root, disk_id);
match fs::read(&path).await {
@@ -156,7 +138,6 @@ pub async fn read_snapshot(root: &Path, disk_id: &str) -> Result<Option<LocalUsa
}
/// Persist a snapshot to disk, creating directories as needed and overwriting any existing file.
#[allow(dead_code, reason = "unwired local usage-snapshot feature; see module docs (backlog#1823)")]
pub async fn write_snapshot(root: &Path, disk_id: &str, snapshot: &LocalUsageSnapshot) -> Result<()> {
let dir = data_usage_dir(root);
fs::create_dir_all(&dir).await.map_err(Error::other)?;
+104 -14
View File
@@ -13,6 +13,7 @@
// limitations under the License.
// #730: scanner/data-usage state is partially migrated and still owns staged cache helpers.
#![allow(dead_code)]
pub mod local_snapshot;
@@ -33,8 +34,8 @@ use crate::{
pub use local_snapshot::{LocalUsageSnapshot, read_snapshot as read_local_snapshot, snapshot_path};
use rustfs_data_usage::{
BucketTargetUsageInfo, BucketUsageInfo, CompressionTotalInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
DataUsageCache, DataUsageInfo, DiskUsageStatus, LEGACY_DATA_USAGE_OBJECT_NAME, SizeHistogram, VersionsHistogram,
observed_data_usage_is_newer,
DataUsageCache, DataUsageEntry, DataUsageInfo, DiskUsageStatus, LEGACY_DATA_USAGE_OBJECT_NAME, SizeHistogram, SizeSummary,
VersionsHistogram, observed_data_usage_is_newer,
};
use rustfs_io_metrics::record_system_path_failure;
use rustfs_utils::path::SLASH_SEPARATOR;
@@ -54,6 +55,7 @@ use tracing::{debug, error, info, instrument};
// Data usage storage constants
pub const DATA_USAGE_ROOT: &str = SLASH_SEPARATOR;
const DATA_COMPRESSION_TOTAL_NAME: &str = ".compression.json";
const DATA_USAGE_BLOOM_NAME: &str = ".bloomcycle.bin";
pub const DATA_USAGE_CACHE_NAME: &str = ".usage-cache.bin";
const DATA_USAGE_CACHE_TTL_SECS: u64 = 30;
const LIVE_BUCKET_USAGE_MAX_ENTRIES: u64 = 1024;
@@ -311,6 +313,11 @@ lazy_static::lazy_static! {
LEGACY_DATA_USAGE_OBJECT_NAME
);
static ref LEGACY_DATA_USAGE_OBJ_BACKUP_PATH: String = format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str());
pub static ref DATA_USAGE_BLOOM_NAME_PATH: String = format!("{}{}{}",
crate::disk::BUCKET_META_PREFIX,
SLASH_SEPARATOR,
DATA_USAGE_BLOOM_NAME
);
pub static ref DATA_COMPRESSION_TOTAL_NAME_PATH: String = format!("{}{}{}",
crate::disk::BUCKET_META_PREFIX,
SLASH_SEPARATOR,
@@ -851,10 +858,6 @@ async fn resolve_loaded_snapshot_pair_with_source(
}
}
#[allow(
dead_code,
reason = "primary/backup snapshot fallback asserted by this file's tests (backlog#1823)"
)]
async fn resolve_loaded_snapshot(
primary: Result<Vec<u8>, Error>,
backup: impl Future<Output = Result<Vec<u8>, Error>>,
@@ -1184,10 +1187,6 @@ pub async fn invalidate_admin_data_usage_snapshot_cache() {
}
/// Aggregate usage information from local disk snapshots.
#[allow(
dead_code,
reason = "reached only through aggregate_local_snapshots, which has no caller (backlog#1823)"
)]
fn merge_snapshot(aggregated: &mut DataUsageInfo, mut snapshot: LocalUsageSnapshot, latest_update: &mut Option<SystemTime>) {
if let Some(update) = snapshot.last_update
&& latest_update.is_none_or(|current| update > current)
@@ -1221,10 +1220,6 @@ fn merge_snapshot(aggregated: &mut DataUsageInfo, mut snapshot: LocalUsageSnapsh
}
}
#[allow(
dead_code,
reason = "entry point of the local usage-snapshot feature, which has had no caller since it landed in #5307 (backlog#1823)"
)]
pub async fn aggregate_local_snapshots(store: Arc<ECStore>) -> Result<(Vec<DiskUsageStatus>, DataUsageInfo), Error> {
let mut aggregated = DataUsageInfo::default();
let mut latest_update: Option<SystemTime> = None;
@@ -1772,6 +1767,11 @@ pub async fn record_bucket_object_write_unknown_previous_memory(bucket: &str, ne
entry.pending_scanner_position = None;
}
/// Fast in-memory increment for immediate quota consistency.
pub async fn increment_bucket_usage_memory(bucket: &str, size_increment: u64) {
record_bucket_object_write_memory(bucket, None, size_increment).await;
}
/// Fast in-memory update for successful object deletes.
pub async fn record_bucket_object_delete_memory(bucket: &str, deleted_size: u64, removed_current_object: bool) {
ensure_bucket_usage_cached(bucket).await;
@@ -1814,6 +1814,11 @@ pub async fn record_bucket_delete_marker_memory(bucket: &str) {
entry.pending_scanner_position = None;
}
/// Fast in-memory decrement for immediate quota consistency
pub async fn decrement_bucket_usage_memory(bucket: &str, size_decrement: u64) {
record_bucket_object_delete_memory(bucket, size_decrement, size_decrement > 0).await;
}
/// Get bucket usage from the authoritative cache for this topology.
async fn get_persisted_bucket_usage(bucket: &str) -> Option<u64> {
let store = runtime_sources::object_store_handle()?;
@@ -2008,6 +2013,91 @@ pub async fn apply_bucket_usage_memory_overlay(data_usage_info: &mut DataUsageIn
apply_bucket_usage_memory_overlay_if_authoritative(data_usage_info, authoritative).await;
}
/// Sync memory cache with backend data (called by scanner)
pub async fn sync_memory_cache_with_backend() -> Result<(), Error> {
if let Some(store) = runtime_sources::object_store_handle() {
match load_data_usage_from_backend(store.clone()).await {
Ok(data_usage_info) => {
replace_bucket_usage_memory_from_info(&data_usage_info).await;
}
Err(e) => {
debug!("Failed to sync memory cache with backend: {}", e);
}
}
}
Ok(())
}
/// Create a data usage cache entry from size summary
pub fn create_cache_entry_from_summary(summary: &SizeSummary) -> DataUsageEntry {
let mut entry = DataUsageEntry::default();
entry.add_sizes(summary);
entry
}
/// Convert data usage cache to DataUsageInfo
pub fn cache_to_data_usage_info(
cache: &DataUsageCache,
path: &str,
buckets: &[crate::storage_api_contracts::bucket::BucketInfo],
) -> DataUsageInfo {
let e = match cache.find(path) {
Some(e) => e,
None => return DataUsageInfo::default(),
};
let flat = cache.flatten(&e);
let mut buckets_usage = HashMap::new();
for bucket in buckets.iter() {
let e = match cache.find(&bucket.name) {
Some(e) => e,
None => continue,
};
let flat = cache.flatten(&e);
let mut bui = BucketUsageInfo {
size: flat.size as u64,
versions_count: flat.versions as u64,
objects_count: flat.objects as u64,
delete_markers_count: flat.delete_markers as u64,
object_size_histogram: flat.obj_sizes.to_map(),
object_versions_histogram: flat.obj_versions.to_map(),
..Default::default()
};
if let Some(rs) = &flat.replication_stats {
bui.replica_size = rs.replica_size;
bui.replica_count = rs.replica_count;
for (arn, stat) in rs.targets.iter() {
bui.replication_info.insert(
arn.clone(),
BucketTargetUsageInfo {
replication_pending_size: stat.pending_size,
replicated_size: stat.replicated_size,
replication_failed_size: stat.failed_size,
replication_pending_count: stat.pending_count,
replication_failed_count: stat.failed_count,
replicated_count: stat.replicated_count,
..Default::default()
},
);
}
}
buckets_usage.insert(bucket.name.clone(), bui);
}
DataUsageInfo {
last_update: cache.info.last_update,
objects_total_count: flat.objects as u64,
versions_total_count: flat.versions as u64,
delete_markers_total_count: flat.delete_markers as u64,
objects_total_size: flat.size as u64,
buckets_count: e.children.len() as u64,
buckets_usage,
..Default::default()
}
}
// Helper functions for DataUsageCache operations
pub async fn load_data_usage_cache(store: &crate::set_disk::SetDisks, name: &str) -> crate::error::Result<DataUsageCache> {
use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
-37
View File
@@ -190,17 +190,6 @@ pub(crate) const GET_METADATA_CACHE_REASON_VERSION_SUSPENDED: &str = "version_su
pub(crate) const GET_METADATA_CACHE_REASON_VERSIONED: &str = "versioned";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA: &str = "conflicting_metadata";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER: &str = "delete_marker";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY: &str = "data_read_inline_body_verify";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED: &str = "data_read_inline_deleted";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY: &str = "data_read_inline_geometry";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH: &str = "data_read_inline_identity_mismatch";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD: &str = "data_read_inline_missing_payload";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD: &str = "data_read_inline_missing_shard";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE: &str = "data_read_inline_not_inline";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE: &str = "data_read_inline_part_shape";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE: &str = "data_read_inline_remote";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE: &str = "data_read_inline_size";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED: &str = "data_read_inline_transformed";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_ERROR: &str = "error";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM: &str = "insufficient_quorum";
pub(crate) const GET_METADATA_EARLY_STOP_REASON_NOT_FOUND: &str = "not_found";
@@ -562,32 +551,6 @@ mod tests {
assert_eq!(GET_METADATA_CACHE_REASON_VERSIONED, "versioned");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA, "conflicting_metadata");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER, "delete_marker");
assert_eq!(
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY,
"data_read_inline_body_verify"
);
assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED, "data_read_inline_deleted");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY, "data_read_inline_geometry");
assert_eq!(
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH,
"data_read_inline_identity_mismatch"
);
assert_eq!(
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD,
"data_read_inline_missing_payload"
);
assert_eq!(
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD,
"data_read_inline_missing_shard"
);
assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE, "data_read_inline_not_inline");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE, "data_read_inline_part_shape");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE, "data_read_inline_remote");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE, "data_read_inline_size");
assert_eq!(
GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED,
"data_read_inline_transformed"
);
assert_eq!(GET_METADATA_EARLY_STOP_REASON_ERROR, "error");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM, "insufficient_quorum");
assert_eq!(GET_METADATA_EARLY_STOP_REASON_NOT_FOUND, "not_found");
+120 -39
View File
@@ -418,6 +418,17 @@ pub struct DiskHealthTracker {
pub last_capacity_free: AtomicU64,
/// Last successful capacity probe timestamp
pub last_capacity_probe_unix_secs: AtomicI64,
/// Authoritative atomically published runtime/status pair.
state_snapshot: AtomicU64,
transition_lock: std::sync::Mutex<()>,
}
fn pack_health_state(runtime_state: RuntimeDriveHealthState, status: u32) -> u64 {
(u64::from(runtime_state as u32) << 32) | u64::from(status)
}
fn unpack_health_state(snapshot: u64) -> (RuntimeDriveHealthState, u32) {
(RuntimeDriveHealthState::from_u32((snapshot >> 32) as u32), snapshot as u32)
}
#[derive(Debug)]
@@ -637,23 +648,14 @@ impl Default for DiskOperationMetrics {
}
impl DiskOperationMetrics {
#[allow(
dead_code,
reason = "internal metrics recorder reached only from record() below (backlog#1823)"
)]
fn record_call(&mut self) {
self.lifetime_calls.fetch_add(1, Ordering::Relaxed);
}
#[allow(
dead_code,
reason = "internal metrics recorder reached only from record() below (backlog#1823)"
)]
fn record_latency(&mut self, now_sec: u64, elapsed: Duration) {
self.record_latency_atomic(now_sec, elapsed);
}
#[allow(dead_code, reason = "metrics roll-up with no caller in this port (backlog#1823)")]
fn record(&mut self, now_sec: u64, elapsed: Duration) {
self.record_call();
self.record_latency(now_sec, elapsed);
@@ -739,6 +741,8 @@ impl DiskHealthTracker {
last_capacity_used: AtomicU64::new(0),
last_capacity_free: AtomicU64::new(0),
last_capacity_probe_unix_secs: AtomicI64::new(0),
state_snapshot: AtomicU64::new(pack_health_state(RuntimeDriveHealthState::Online, DISK_HEALTH_OK)),
transition_lock: std::sync::Mutex::new(()),
}
}
@@ -775,39 +779,56 @@ impl DiskHealthTracker {
/// Check if disk is faulty
pub fn is_faulty(&self) -> bool {
self.status.load(Ordering::Acquire) == DISK_HEALTH_FAULTY
unpack_health_state(self.state_snapshot.load(Ordering::Acquire)).1 == DISK_HEALTH_FAULTY
}
pub fn health_state_snapshot(&self) -> (RuntimeDriveHealthState, bool) {
let (runtime_state, status) = unpack_health_state(self.state_snapshot.load(Ordering::Acquire));
(runtime_state, status == DISK_HEALTH_FAULTY)
}
fn publish_state(&self, runtime_state: RuntimeDriveHealthState, status: u32) {
self.state_snapshot
.store(pack_health_state(runtime_state, status), Ordering::Release);
self.runtime_state.store(runtime_state as u32, Ordering::Release);
self.status.store(status, Ordering::Release);
}
/// Set disk as faulty
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn set_faulty(&self) {
self.status.store(DISK_HEALTH_FAULTY, Ordering::Release);
let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
self.publish_state(RuntimeDriveHealthState::Offline, DISK_HEALTH_FAULTY);
}
/// Set disk as OK
pub fn set_ok(&self) {
self.status.store(DISK_HEALTH_OK, Ordering::Release);
let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
self.publish_state(RuntimeDriveHealthState::Online, DISK_HEALTH_OK);
}
#[cfg(test)]
pub fn force_runtime_state_for_test(&self, state: RuntimeDriveHealthState) {
self.runtime_state.store(state as u32, Ordering::Release);
match state {
RuntimeDriveHealthState::Offline => self.set_faulty(),
RuntimeDriveHealthState::Online | RuntimeDriveHealthState::Suspect | RuntimeDriveHealthState::Returning => {
self.set_ok();
}
}
let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
let status = if state == RuntimeDriveHealthState::Offline {
DISK_HEALTH_FAULTY
} else {
DISK_HEALTH_OK
};
self.publish_state(state, status);
}
pub fn swap_ok_to_faulty(&self) -> bool {
self.status
.compare_exchange(DISK_HEALTH_OK, DISK_HEALTH_FAULTY, Ordering::AcqRel, Ordering::Relaxed)
.is_ok()
let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
let (_, status) = unpack_health_state(self.state_snapshot.load(Ordering::Acquire));
if status != DISK_HEALTH_OK {
return false;
}
self.publish_state(RuntimeDriveHealthState::Offline, DISK_HEALTH_FAULTY);
true
}
pub fn runtime_state(&self) -> RuntimeDriveHealthState {
RuntimeDriveHealthState::from_u32(self.runtime_state.load(Ordering::Acquire))
unpack_health_state(self.state_snapshot.load(Ordering::Acquire)).0
}
pub fn offline_duration(&self) -> Option<Duration> {
@@ -823,6 +844,7 @@ impl DiskHealthTracker {
}
pub fn mark_failure(&self, endpoint: &Endpoint, reason: &'static str) -> bool {
let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
let current = self.runtime_state();
let now = current_unix_secs();
let next = match current {
@@ -851,24 +873,18 @@ impl DiskHealthTracker {
};
let became_offline = next == RuntimeDriveHealthState::Offline && current != RuntimeDriveHealthState::Offline;
if next == RuntimeDriveHealthState::Offline {
self.status.store(DISK_HEALTH_FAULTY, Ordering::Release);
} else {
self.status.store(DISK_HEALTH_OK, Ordering::Release);
}
self.transition_state(endpoint, current, next, reason);
became_offline
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn mark_offline(&self, endpoint: &Endpoint, reason: &'static str) -> bool {
let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
let current = self.runtime_state();
if current == RuntimeDriveHealthState::Offline {
return false;
}
self.consecutive_successes.store(0, Ordering::Release);
self.status.store(DISK_HEALTH_FAULTY, Ordering::Release);
self.transition_state(endpoint, current, RuntimeDriveHealthState::Offline, reason);
true
}
@@ -882,11 +898,10 @@ impl DiskHealthTracker {
}
fn reset_for_store_init_retry_at(&self, endpoint: &Endpoint, now: Duration) {
let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
let now_nanos = unix_nanos(now);
let now_secs = unix_secs_i64(now);
self.status.store(DISK_HEALTH_OK, Ordering::Release);
self.runtime_state
.store(RuntimeDriveHealthState::Online as u32, Ordering::Release);
self.publish_state(RuntimeDriveHealthState::Online, DISK_HEALTH_OK);
self.consecutive_failures.store(0, Ordering::Release);
self.consecutive_successes.store(0, Ordering::Release);
self.offline_since_unix_secs.store(0, Ordering::Release);
@@ -898,6 +913,7 @@ impl DiskHealthTracker {
}
pub fn mark_recovery_success(&self, endpoint: &Endpoint, reason: &'static str) -> bool {
let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
let current = self.runtime_state();
let next = match current {
RuntimeDriveHealthState::Online => RuntimeDriveHealthState::Online,
@@ -918,7 +934,6 @@ impl DiskHealthTracker {
let became_online = next == RuntimeDriveHealthState::Online;
if became_online {
self.status.store(DISK_HEALTH_OK, Ordering::Release);
self.consecutive_failures.store(0, Ordering::Release);
self.consecutive_successes.store(0, Ordering::Release);
}
@@ -948,7 +963,13 @@ impl DiskHealthTracker {
return;
}
self.runtime_state.store(next as u32, Ordering::Release);
let current_status = unpack_health_state(self.state_snapshot.load(Ordering::Acquire)).1;
let status = match next {
RuntimeDriveHealthState::Offline => DISK_HEALTH_FAULTY,
RuntimeDriveHealthState::Returning => current_status,
RuntimeDriveHealthState::Online | RuntimeDriveHealthState::Suspect => DISK_HEALTH_OK,
};
self.publish_state(next, status);
self.last_transition_unix_secs
.store(current_unix_secs() as i64, Ordering::Release);
@@ -991,13 +1012,11 @@ impl DiskHealthTracker {
}
/// Get waiting operations count
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn waiting_count(&self) -> u32 {
self.waiting.load(Ordering::Relaxed)
}
/// Get last success timestamp
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn last_success(&self) -> i64 {
self.last_success.load(Ordering::Acquire)
}
@@ -1039,6 +1058,21 @@ impl Default for DiskHealthTracker {
}
}
/// Health check context key for tracking disk operations
#[derive(Debug, Clone)]
struct HealthDiskCtxKey;
#[derive(Debug)]
struct HealthDiskCtxValue {
last_success: Arc<AtomicI64>,
}
impl HealthDiskCtxValue {
fn log_success(&self) {
self.last_success.store(current_unix_nanos(), Ordering::Relaxed);
}
}
/// LocalDiskWrapper wraps a DiskStore with health tracking capabilities.
/// This is similar to Go's xlStorageDiskIDCheck.
#[derive(Debug, Clone)]
@@ -1070,6 +1104,10 @@ impl LocalDiskWrapper {
)
}
pub(crate) fn new_with_health(disk: Arc<LocalDisk>, health_check: bool, health: Arc<DiskHealthTracker>) -> Self {
Self::new_with_health_and_metrics(disk, health_check, health, Arc::new(DiskHealthMetricEpoch::default()))
}
pub(crate) fn new_with_reconnect_state(
disk: Arc<LocalDisk>,
health_check: bool,
@@ -1217,7 +1255,7 @@ impl LocalDiskWrapper {
return;
}
if health.status.load(Ordering::Relaxed) != DISK_HEALTH_OK {
if health.is_faulty() {
continue;
}
@@ -1432,6 +1470,20 @@ impl LocalDiskWrapper {
}
}
async fn check_id(&self, want_id: Option<Uuid>) -> Result<()> {
if want_id.is_none() {
return Ok(());
}
let stored_disk_id = self.disk.get_disk_id().await?;
if stored_disk_id != want_id {
return Err(Error::other(format!("Disk ID mismatch wanted {want_id:?}, got {stored_disk_id:?}")));
}
Ok(())
}
/// Check if disk ID is stale
async fn check_disk_stale(&self) -> Result<()> {
let Some(current_disk_id) = *self.disk_id.read().await else {
@@ -2909,6 +2961,35 @@ mod tests {
});
}
#[test]
fn concurrent_failure_and_recovery_publish_one_health_snapshot() {
let endpoint = Endpoint::try_from("/tmp/concurrent-health-snapshot").expect("endpoint should parse");
let health = Arc::new(DiskHealthTracker::new());
let workers = (0..8)
.map(|_| {
let health = Arc::clone(&health);
let endpoint = endpoint.clone();
std::thread::spawn(move || {
for _ in 0..32 {
health.mark_failure(&endpoint, "concurrent_test");
health.mark_recovery_success(&endpoint, "concurrent_test");
let (runtime, faulty) = health.health_state_snapshot();
assert!(matches!(
(runtime, faulty),
(RuntimeDriveHealthState::Online, false)
| (RuntimeDriveHealthState::Suspect, false)
| (RuntimeDriveHealthState::Offline, true)
| (RuntimeDriveHealthState::Returning, true)
));
}
})
})
.collect::<Vec<_>>();
for worker in workers {
worker.join().expect("health transition worker should not panic");
}
}
#[test]
fn operation_success_recovers_suspect_drive_without_faulting() {
let endpoint = Endpoint::try_from("/tmp/runtime-state-suspect-success").expect("endpoint should parse");
-1
View File
@@ -48,7 +48,6 @@ pub fn to_volume_error(io_err: std::io::Error) -> std::io::Error {
}
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn to_disk_error(io_err: std::io::Error) -> std::io::Error {
match io_err.kind() {
std::io::ErrorKind::NotFound => DiskError::DiskNotFound.into(),
-1
View File
@@ -178,7 +178,6 @@ pub async fn remove(path: impl AsRef<Path>) -> io::Result<()> {
}
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub async fn remove_all(path: impl AsRef<Path>) -> io::Result<()> {
// Try remove_file first; fall back to remove_dir_all if it's a directory
match fs::remove_file(path.as_ref()).await {
+1 -70
View File
@@ -665,7 +665,6 @@ async fn remove_empty_directory_tree_under_mount_lease(
}
#[cfg(unix)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
async fn remove_empty_directory_tree_with(
root: &Path,
before_descend: impl FnMut(&Path) -> std::io::Result<()>,
@@ -1017,29 +1016,13 @@ fn record_direct_read_page_fault_delta(path: &'static str, stage: &'static str,
/// When enabled, shard reads bypass the page cache using O_DIRECT flag.
/// Requires aligned buffers (typically 512 bytes or 4096 bytes).
/// Default: false (uses page cache via mmap/pread).
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE: &str = "RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE";
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE: bool = false;
/// Minimum shard size threshold for O_DIRECT reads.
/// Only shards larger than this threshold will use O_DIRECT.
/// Default: 4MB.
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: &str = "RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD";
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: usize = 4 * 1024 * 1024;
/// Enable O_DIRECT for erasure shard / multipart part data writes (Linux only).
@@ -1053,15 +1036,7 @@ const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: usize = 4 * 1024 * 1024;
/// EINVAL/EOPNOTSUPP (tmpfs, overlayfs, 9p, ...) latch the path off and fall
/// back to buffered writes for the whole disk. Non-Linux always falls back.
/// Default: false (buffered writes via the page cache, as before).
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: &str = "RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE";
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: bool = false;
const ENV_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: &str = "RUSTFS_OBJECT_MMAP_POPULATE_ENABLE";
const DEFAULT_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: bool = false;
@@ -1120,14 +1095,12 @@ macro_rules! cached_read_env {
cached_read_env! {
/// Check if O_DIRECT reads are enabled.
#[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
fn is_direct_io_read_enabled() -> bool =
rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE);
}
cached_read_env! {
/// Check if O_DIRECT shard/part data writes are enabled.
#[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
fn is_direct_io_write_enabled() -> bool =
rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE);
}
@@ -1483,7 +1456,6 @@ pub(crate) fn effective_durability(volume: &str) -> DurabilityMode {
cached_read_env! {
/// Get the O_DIRECT read threshold size.
#[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
fn get_direct_io_read_threshold() -> usize =
rustfs_utils::get_env_usize(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD);
}
@@ -1701,20 +1673,12 @@ impl DirectIoWriteState {
/// Target staging size for O_DIRECT writes, rounded up to the DIO alignment.
/// Bounds the per-writer aligned bounce buffer and batches many shard blocks
/// into one positioned write to keep the syscall count low.
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const DIRECT_WRITE_STAGING_BYTES: usize = 1024 * 1024;
/// Aligned bounce-buffer capacity for a given DIO alignment: the target staging
/// size rounded up to a whole multiple of `align` so the buffer address, every
/// flushed batch length, and every write offset stay alignment-correct.
/// Platform-independent (no O_DIRECT), so it is unit-tested on any host.
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
fn direct_write_staging_capacity(align: usize) -> usize {
debug_assert!(align.is_power_of_two() && align >= 512);
DIRECT_WRITE_STAGING_BYTES.div_ceil(align) * align
@@ -1723,10 +1687,6 @@ fn direct_write_staging_capacity(align: usize) -> usize {
/// Split `filled` staged bytes into the alignment-sized prefix written with
/// O_DIRECT and the sub-alignment tail written buffered. Platform-independent,
/// so the tail-boundary math is unit-tested on any host.
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
fn direct_write_tail_split(filled: usize, align: usize) -> (usize, usize) {
let aligned = filled - (filled % align);
(aligned, filled - aligned)
@@ -2182,7 +2142,6 @@ fn set_delete_version_fail_after_data_staged(path: &str) {
}
#[cfg(test)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(crate) fn set_delete_version_fail_after_commit(root: &Path, path: &str) {
DELETE_VERSION_FAIL_AFTER_COMMIT
.lock()
@@ -2488,10 +2447,6 @@ enum SyncMode {
FileOnly,
}
#[allow(
dead_code,
reason = "reclaim bookkeeping fields written by Drop but never read back (backlog#1823)"
)]
struct FileCacheReclaimWriter {
inner: File,
reclaim_len: usize,
@@ -2499,10 +2454,6 @@ struct FileCacheReclaimWriter {
reclaimed: bool,
}
#[allow(
dead_code,
reason = "reclaim bookkeeping fields written by Drop but never read back (backlog#1823)"
)]
struct FileCacheReclaimReader {
inner: File,
reclaim_offset: u64,
@@ -2568,10 +2519,6 @@ impl<R: AsyncRead + Unpin> AsyncRead for StallTimeoutReader<R> {
}
}
#[allow(
dead_code,
reason = "reclaim metrics emitter reached only from the Linux-gated reclaim paths (backlog#1823)"
)]
fn record_file_cache_reclaim_success(kind: &'static str, reclaim_len: usize, started: std::time::Instant) {
// Runs per read-stream page-cache reclaim window; skip the whole emission
// (three metric-key constructions) when general metrics are disabled.
@@ -3124,7 +3071,6 @@ impl LocalIoBackend for StdBackend {
use memmap2::MmapOptions;
use std::time::{Duration as StdDuration, Instant as StdInstant};
#[allow(dead_code, reason = "mmap copy result slot kept beside the mapping it owns (backlog#1823)")]
struct MmapCopyReadResult {
bytes: Bytes,
access_check_duration: StdDuration,
@@ -4758,10 +4704,6 @@ fn build_local_io_backend(root: PathBuf) -> Arc<dyn LocalIoBackend> {
Arc::new(StdBackend::new(root))
}
#[allow(
dead_code,
reason = "path cache and cwd slots retained beside the disk root they derive from (backlog#1823)"
)]
pub struct LocalDisk {
pub root: PathBuf,
publication_root: os::PublicationRoot,
@@ -5548,7 +5490,6 @@ impl LocalDisk {
Ok(Self::resolve_abs_path_from(&self.root, path.as_ref()))
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn io_resolve_abs_path(&self, path: impl AsRef<Path>) -> PathBuf {
let path_ref = path.as_ref();
let path_str = path_ref.to_string_lossy();
@@ -5626,24 +5567,15 @@ impl LocalDisk {
}
// Check if a path is valid
#[allow(
dead_code,
reason = "method wrapper over the live free function check_local_disk_valid_path; no caller in this port (backlog#1823)"
)]
fn check_valid_path<P: AsRef<Path>>(&self, path: P) -> Result<()> {
check_local_disk_valid_path(self.io_root(), path)
}
#[allow(
dead_code,
reason = "method wrapper over the live free function reject_local_disk_symlink_components; no caller in this port (backlog#1823)"
)]
fn reject_symlink_components(&self, path: &Path) -> Result<()> {
reject_local_disk_symlink_components(self.io_root(), path)
}
// Batch path generation with single lock acquisition
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn get_object_paths_batch(&self, requests: &[(String, String)]) -> Result<Vec<PathBuf>> {
let mut results = Vec::with_capacity(requests.len());
let mut cache_misses = Vec::new();
@@ -6556,13 +6488,12 @@ impl LocalDisk {
Ok(f)
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
async fn open_file_read_only(&self, path: impl AsRef<Path>) -> Result<File> {
let f = super::fs::open_file(path.as_ref(), O_RDONLY).await.map_err(to_file_error)?;
Ok(f)
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
#[allow(dead_code)]
fn get_metrics(&self) -> DiskMetrics {
DiskMetrics::default()
}
+1 -5
View File
@@ -13,6 +13,7 @@
// limitations under the License.
// #730: disk abstractions still carry staged health and direct-I/O migration paths.
#![allow(dead_code)]
pub mod disk_store;
pub mod endpoint;
@@ -1113,10 +1114,6 @@ pub struct DiskInfo {
}
#[derive(Clone, Debug, Default)]
#[allow(
dead_code,
reason = "MinIO-parity disk info shape with no constructor in this port (backlog#1823)"
)]
pub struct Info {
pub total: u64,
pub free: u64,
@@ -1375,7 +1372,6 @@ pub fn conv_part_err_to_int(err: &Option<Error>) -> usize {
}
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn has_part_err(part_errs: &[usize]) -> bool {
part_errs.iter().any(|err| *err != CHECK_PART_SUCCESS)
}
+11 -5
View File
@@ -571,10 +571,6 @@ fn regular_files(dir: &Path) -> io::Result<Vec<PathBuf>> {
/// Fdatasync every regular file directly inside `dir`, then fsync the directory
/// itself.
#[allow(
dead_code,
reason = "reached only through sync_dir_files, whose callers are tests (backlog#1823)"
)]
pub fn sync_dir_files_std(dir: impl AsRef<Path>) -> io::Result<()> {
for entry in std::fs::read_dir(dir.as_ref())? {
let entry = entry?;
@@ -587,7 +583,6 @@ pub fn sync_dir_files_std(dir: impl AsRef<Path>) -> io::Result<()> {
/// Async wrapper around [`sync_dir_files_std`]. Large directories flush files
/// concurrently, bounded both per directory and process-wide.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub async fn sync_dir_files(dir: impl AsRef<Path>) -> io::Result<()> {
sync_dir_files_with_limiter(dir, Arc::new(Semaphore::new(MAX_PARALLEL_FILE_SYNCS))).await
}
@@ -1814,6 +1809,10 @@ impl RenameCommitGuard {
})
}
pub(crate) fn lock_destination_directory_for_path_access(&self, directory: &Path) -> io::Result<RenameDestinationPathGuard> {
self.destination_directory_guard(directory, false)
}
pub(crate) fn create_destination_directory_for_path_access(
&self,
directory: &Path,
@@ -2859,6 +2858,13 @@ pub async fn os_mkdir_all(dir_path: impl AsRef<Path>, base_dir: impl AsRef<Path>
Ok(())
}
/// Check if a file exists.
/// Returns true if the file exists, false otherwise.
#[tracing::instrument(level = "debug", skip_all)]
pub fn file_exists(path: impl AsRef<Path>) -> bool {
std::fs::metadata(path.as_ref()).map(|_| true).unwrap_or(false)
}
/// Whether an [`io::Error`] means "the directory is not empty".
///
/// POSIX lets `rmdir`/`rename` report a non-empty directory as either
+12 -291
View File
@@ -820,263 +820,10 @@ impl BitrotWriterWrapper {
}
}
// --- startup bitrot self-test (rustfs/backlog#1873, MinIO bitrotSelfTest parity) ---
//
// A broken hash implementation (bad SIMD feature combination, platform drift, a
// key-handling regression) fails silently: every shard reads back "corrupt",
// heal rewrites data that was fine, and cross-platform clusters disagree about
// which copy is healthy. The self-test below pins the algorithms the moment a
// process starts, so a drifted build announces itself instead of quietly
// rewriting objects. See docs/rustfs-heal-scanner-vs-minio-comprehensive-
// analysis-2026-08-16.md §6 HS-11.
/// Length of the deterministic self-test payload.
pub const BITROT_SELF_TEST_PAYLOAD_LEN: usize = 4096;
/// Known-answer digest of [`bitrot_self_test_payload`] under `HighwayHash256S`
/// (the production default). Pinned so any platform or build where the
/// implementation drifts fails startup instead of miss-hashing shards.
const BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S: [u8; 32] = [
0xb9, 0x32, 0xa2, 0xaa, 0x4a, 0xb7, 0x33, 0x6a, 0xa3, 0xca, 0x7e, 0x61, 0x9d, 0x86, 0x52, 0x14, 0x6e, 0x7f, 0xd8, 0x9e, 0xea,
0x08, 0xd9, 0x8c, 0x33, 0x85, 0x87, 0x19, 0x30, 0xd6, 0xed, 0x06,
];
/// Known-answer digest of the same payload under `HighwayHash256SLegacy`.
const BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S_LEGACY: [u8; 32] = [
0x98, 0x24, 0x71, 0x4f, 0x16, 0xbb, 0x48, 0x39, 0xed, 0x68, 0xfa, 0x63, 0x5e, 0xd9, 0x07, 0x61, 0xdf, 0x0a, 0xff, 0xcf, 0x7d,
0x8c, 0xa8, 0xc7, 0xc0, 0xb6, 0x6f, 0x05, 0xdb, 0xda, 0x5a, 0x22,
];
/// FIPS 180-2 test vector: SHA-256 of the ASCII string "abc". Unlike the
/// Highway digests above this one is externally verifiable, so it guards the
/// whole `HashAlgorithm` plumbing even for readers who distrust pinned
/// self-computed constants.
const BITROT_SELF_TEST_KAT_SHA256_ABC: [u8; 32] = [
0xba, 0x78, 0x16, 0xbf, 0x8f, 0x01, 0xcf, 0xea, 0x41, 0x41, 0x40, 0xde, 0x5d, 0xae, 0x22, 0x23, 0xb0, 0x03, 0x61, 0xa3, 0x96,
0x17, 0x7a, 0x9c, 0xb4, 0x10, 0xff, 0x61, 0xf2, 0x00, 0x15, 0xad,
];
/// Deterministic self-test payload: xorshift64* from a fixed seed, so every
/// platform and every run hashes the same 4096 bytes.
fn bitrot_self_test_payload() -> [u8; BITROT_SELF_TEST_PAYLOAD_LEN] {
let mut state = 0x9E37_79B9_7F4A_7C15u64;
let mut payload = [0u8; BITROT_SELF_TEST_PAYLOAD_LEN];
for byte in payload.iter_mut() {
state ^= state >> 12;
state ^= state << 25;
state ^= state >> 27;
*byte = state.wrapping_mul(0x2545_F491_4F6C_DD1D) as u8;
}
payload
}
/// Why a bitrot self-test failed.
#[derive(Debug)]
pub enum BitrotSelfTestError {
/// A known-answer digest mismatched the pinned constant.
KnownAnswerMismatch {
algorithm: &'static str,
got: String,
want: String,
},
/// A freshly encoded shard failed `bitrot_verify`.
RoundtripVerify { algorithm: &'static str, detail: String },
/// A verified roundtrip read back different bytes than were written.
RoundtripReadback { algorithm: &'static str },
/// A deliberately tampered shard was not rejected by `bitrot_verify`.
TamperNotRejected {
algorithm: &'static str,
tampered: &'static str,
},
}
impl std::fmt::Display for BitrotSelfTestError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::KnownAnswerMismatch { algorithm, got, want } => {
write!(f, "known-answer mismatch for {algorithm}: got {got}, want {want}")
}
Self::RoundtripVerify { algorithm, detail } => write!(f, "{algorithm} roundtrip shard failed verification: {detail}"),
Self::RoundtripReadback { algorithm } => write!(f, "{algorithm} roundtrip read back different bytes"),
Self::TamperNotRejected { algorithm, tampered } => {
write!(f, "{algorithm} tampered shard ({tampered}) was not rejected")
}
}
}
}
impl std::error::Error for BitrotSelfTestError {}
fn self_test_hex(bytes: &[u8]) -> String {
rustfs_utils::hex(bytes)
}
// (kept as a named one-liner so every KAT failure site reads the same; the
// underlying formatter is the shared `rustfs_utils::hex`)
/// Compare a digest against its pinned constant. Split out so a test can drive
/// it with a wrong constant and prove the mismatch path fires.
fn bitrot_kat_check(
algorithm: &'static str,
algo: &HashAlgorithm,
payload: &[u8],
expected: &[u8; 32],
) -> Result<(), BitrotSelfTestError> {
let digest = algo.hash_encode(payload);
let digest = digest.as_ref();
if digest.len() != expected.len() || digest != expected.as_slice() {
return Err(BitrotSelfTestError::KnownAnswerMismatch {
algorithm,
got: self_test_hex(digest),
want: self_test_hex(expected),
});
}
Ok(())
}
/// Encode `payload` with `shard_size` blocks, verify it end to end, and read
/// every block back through `BitrotReader` comparing bytes.
async fn bitrot_roundtrip_check(
algorithm: &'static str,
algo: HashAlgorithm,
payload: &[u8],
shard_size: usize,
) -> Result<(), BitrotSelfTestError> {
let mut writer = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
for chunk in payload.chunks(shard_size) {
writer
.write(chunk)
.await
.map_err(|err| BitrotSelfTestError::RoundtripVerify {
algorithm,
detail: format!("encode failed: {err}"),
})?;
}
let encoded = writer.into_inner().into_inner();
let on_disk = bitrot_shard_file_size(payload.len(), shard_size, algo.clone());
if encoded.len() != on_disk {
return Err(BitrotSelfTestError::RoundtripVerify {
algorithm,
detail: format!("encoded {} bytes, size formula says {on_disk}", encoded.len()),
});
}
bitrot_verify(std::io::Cursor::new(encoded.clone()), on_disk, payload.len(), algo.clone(), shard_size)
.await
.map_err(|err| BitrotSelfTestError::RoundtripVerify {
algorithm,
detail: err.to_string(),
})?;
let mut reader = BitrotReader::new(std::io::Cursor::new(encoded), shard_size, algo, false);
let mut offset = 0usize;
while offset < payload.len() {
let want = shard_size.min(payload.len() - offset);
let mut buf = vec![0u8; want];
let read = reader
.read(&mut buf)
.await
.map_err(|err| BitrotSelfTestError::RoundtripVerify {
algorithm,
detail: format!("read back failed at offset {offset}: {err}"),
})?;
if read != want || buf[..read] != payload[offset..offset + read] {
return Err(BitrotSelfTestError::RoundtripReadback { algorithm });
}
offset += read;
}
Ok(())
}
/// Flip one byte and require `bitrot_verify` to reject the result.
async fn bitrot_tamper_check(
algorithm: &'static str,
algo: HashAlgorithm,
payload: &[u8],
shard_size: usize,
tampered: &'static str,
flip_at: usize,
) -> Result<(), BitrotSelfTestError> {
let mut writer = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
for chunk in payload.chunks(shard_size) {
writer.write(chunk).await.expect("self-test encode should not fail");
}
let mut corrupt = writer.into_inner().into_inner();
let flip_index = flip_at % corrupt.len();
corrupt[flip_index] ^= 0x80;
let on_disk = bitrot_shard_file_size(payload.len(), shard_size, algo.clone());
match bitrot_verify(std::io::Cursor::new(corrupt), on_disk, payload.len(), algo, shard_size).await {
// The flipped byte must be rejected as a hash mismatch specifically, not
// by any incidental read error: an in-memory cursor cannot fail reads,
// so accepting any other failure here would mask a verify path that
// errors out before it ever compares hashes.
Err(err) if err.to_string().contains("hash mismatch") => Ok(()),
Ok(()) => Err(BitrotSelfTestError::TamperNotRejected { algorithm, tampered }),
Err(err) => Err(BitrotSelfTestError::RoundtripVerify {
algorithm,
detail: format!("tampered shard rejected with an unexpected error: {err}"),
}),
}
}
/// Verify every bitrot algorithm this crate can write or verify in production:
/// both streaming Highway variants roundtrip end to end (encode → size formula
/// → `bitrot_verify` → read back) and reject a flipped byte in both the data
/// and the leading hash, while all three hashed algorithms reproduce their
/// pinned known-answer digests.
///
/// Runs in well under a millisecond on 4 KiB of data; callers may run it inline
/// at startup. Pure CPU, no allocation beyond a few KiB of scratch.
pub async fn bitrot_self_test() -> Result<(), BitrotSelfTestError> {
let payload = bitrot_self_test_payload();
// Externally verifiable vector first: it guards the HashAlgorithm plumbing
// itself, before any self-pinned constants are consulted.
let abc = HashAlgorithm::SHA256.hash_encode(b"abc");
if abc.as_ref() != BITROT_SELF_TEST_KAT_SHA256_ABC.as_slice() {
return Err(BitrotSelfTestError::KnownAnswerMismatch {
algorithm: "SHA256",
got: self_test_hex(abc.as_ref()),
want: self_test_hex(&BITROT_SELF_TEST_KAT_SHA256_ABC),
});
}
bitrot_kat_check(
"HighwayHash256S",
&HashAlgorithm::HighwayHash256S,
&payload,
&BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S,
)?;
bitrot_kat_check(
"HighwayHash256SLegacy",
&HashAlgorithm::HighwayHash256SLegacy,
&payload,
&BITROT_SELF_TEST_KAT_HIGHWAY_HASH256S_LEGACY,
)?;
for (algorithm, algo) in [
("HighwayHash256S", HashAlgorithm::HighwayHash256S),
("HighwayHash256SLegacy", HashAlgorithm::HighwayHash256SLegacy),
] {
// Full blocks plus a partial tail, exactly like a real part stripe.
let tail_len = 2 * 1024 + 333;
bitrot_roundtrip_check(algorithm, algo.clone(), &payload, 1024).await?;
bitrot_roundtrip_check(algorithm, algo.clone(), &payload[..tail_len], 1024).await?;
// One flipped byte in the final data block, one in the first leading
// hash: both must fail verification.
bitrot_tamper_check(algorithm, algo.clone(), &payload, 1024, "final data byte", payload.len() - 1).await?;
bitrot_tamper_check(algorithm, algo, &payload, 1024, "leading hash byte", 0).await?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::{
BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, bitrot_kat_check, bitrot_self_test,
bitrot_self_test_payload, bitrot_shard_file_size, bitrot_verify, write_all_vectored,
BitrotReader, BitrotWriter, BitrotWriterWrapper, CustomWriter, bitrot_shard_file_size, bitrot_verify, write_all_vectored,
};
use super::{MAX_RETAINED_CHUNKS_PER_BLOCK, ShardChunkRead, ShardSource};
use bytes::Bytes;
@@ -1343,32 +1090,6 @@ mod tests {
}
}
#[test]
fn bitrot_self_test_payload_is_deterministic() {
// Two independent builds of the payload must agree byte for byte, or
// the pinned known-answer digests below would be meaningless.
assert_eq!(bitrot_self_test_payload(), bitrot_self_test_payload());
}
#[test]
fn bitrot_self_test_rejects_a_wrong_known_answer_digest() {
let payload = bitrot_self_test_payload();
let wrong = [0u8; 32];
let err = bitrot_kat_check("HighwayHash256S", &HashAlgorithm::HighwayHash256S, &payload, &wrong)
.expect_err("a zeroed digest must never match");
match err {
super::BitrotSelfTestError::KnownAnswerMismatch { algorithm, .. } => assert_eq!(algorithm, "HighwayHash256S"),
other => panic!("expected KnownAnswerMismatch, got {other:?}"),
}
}
#[tokio::test]
async fn bitrot_self_test_passes() {
bitrot_self_test()
.await
.expect("the pinned digests and roundtrip checks must all pass on this platform");
}
#[tokio::test]
async fn vectored_test_writers_cover_fallback_flush_and_shutdown_paths() {
let mut counting = VectoredCountingWriter::default();
@@ -1468,7 +1189,7 @@ mod tests {
let last = corrupt.len() - 1;
corrupt[last] ^= 0x80;
let err = bitrot_verify(
std::io::Cursor::new(corrupt),
Cursor::new(corrupt),
super::bitrot_shard_file_size(data.len(), shard_size, algo.clone()),
data.len(),
algo,
@@ -1561,7 +1282,7 @@ mod tests {
#[tokio::test]
async fn bitrot_reader_rejects_output_buffers_larger_than_shard_size() {
let mut reader = BitrotReader::new(std::io::Cursor::new(Vec::<u8>::new()), 4, HashAlgorithm::None, false);
let mut reader = BitrotReader::new(Cursor::new(Vec::<u8>::new()), 4, HashAlgorithm::None, false);
let mut out = [0u8; 5];
let err = reader
.read(&mut out)
@@ -1686,7 +1407,7 @@ mod tests {
(HashAlgorithm::HighwayHash256, true),
] {
let label = format!("{algo:?}");
let writer = std::io::Cursor::new(Vec::<u8>::new());
let writer = Cursor::new(Vec::<u8>::new());
let mut w = BitrotWriter::new(writer, shard_size, algo.clone());
w.write(&[7u8; 16]).await.unwrap();
let written = w.into_inner().into_inner();
@@ -1771,7 +1492,7 @@ mod tests {
}
async fn encode_one_block(payload: &[u8], shard_size: usize, algo: HashAlgorithm) -> Vec<u8> {
let mut w = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo);
let mut w = BitrotWriter::new(Cursor::new(Vec::<u8>::new()), shard_size, algo);
w.write(payload).await.unwrap();
w.into_inner().into_inner()
}
@@ -1879,7 +1600,7 @@ mod tests {
for algo in [HashAlgorithm::HighwayHash256S, HashAlgorithm::HighwayHash256SLegacy] {
for &size in &[1usize, 16, 17, 32, 40, 48] {
let payload: Vec<u8> = (0..size).map(|i| i as u8).collect();
let mut w = BitrotWriter::new(std::io::Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
let mut w = BitrotWriter::new(Cursor::new(Vec::<u8>::new()), shard_size, algo.clone());
for chunk in payload.chunks(shard_size) {
w.write(chunk).await.unwrap();
}
@@ -1953,14 +1674,14 @@ mod tests {
w.write(&data).await.expect("write shard");
let mut via_read = vec![0u8; SHARD];
let n1 = BitrotReader::new(std::io::Cursor::new(encoded.clone()), SHARD, algo.clone(), false)
let n1 = BitrotReader::new(Cursor::new(encoded.clone()), SHARD, algo.clone(), false)
.read(&mut via_read)
.await
.expect("read");
// A buffer with only capacity — no initialized bytes at all.
let mut via_append: Vec<u8> = Vec::with_capacity(SHARD);
let n2 = BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo.clone(), false)
let n2 = BitrotReader::new(Cursor::new(encoded), SHARD, algo.clone(), false)
.read_appending(&mut via_append, SHARD)
.await
.expect("read_appending");
@@ -1985,7 +1706,7 @@ mod tests {
encoded.truncate(encoded.len() - 1);
let mut out: Vec<u8> = Vec::with_capacity(SHARD);
let err = BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo.clone(), false)
let err = BitrotReader::new(Cursor::new(encoded), SHARD, algo.clone(), false)
.read_appending(&mut out, SHARD)
.await
.expect_err("a truncated shard must not succeed");
@@ -2011,7 +1732,7 @@ mod tests {
encoded[last] ^= 0xff;
let mut out: Vec<u8> = Vec::with_capacity(SHARD);
let err = BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo, false)
let err = BitrotReader::new(Cursor::new(encoded), SHARD, algo, false)
.read_appending(&mut out, SHARD)
.await
.expect_err("a corrupt shard must not verify");
@@ -2123,7 +1844,7 @@ mod tests {
"Cursor<Bytes> must be able to hand out a block, otherwise the fast path is dead code"
);
assert_eq!(mem.position(), 8, "taking a block must advance like a read of the same length");
let mut streamed = std::io::Cursor::new(encoded.clone());
let mut streamed = Cursor::new(encoded.clone());
assert!(
ShardSource::try_take_block(&mut streamed, 8).is_none(),
"a non-Bytes source must stay on the streaming path"
@@ -2151,7 +1872,7 @@ mod tests {
);
let mut via_stream: Vec<u8> = Vec::with_capacity(SHARD);
BitrotReader::new(std::io::Cursor::new(encoded), SHARD, algo, false)
BitrotReader::new(Cursor::new(encoded), SHARD, algo, false)
.read_appending(&mut via_stream, SHARD)
.await
.expect("streaming read");
+69
View File
@@ -1075,6 +1075,9 @@ pub struct GenericError {
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum ObjectApiError {
#[error("BackendDown")]
BackendDown(String),
#[error("The operation is not valid for the current state of the object {}/{}({})", .0.bucket, .0.object, .0.version_id)]
InvalidObjectState(GenericError),
}
@@ -1091,6 +1094,72 @@ pub struct ErrorResponse {
pub host_id: String,
}
pub fn error_resp_to_object_err(err: ErrorResponse, params: Vec<&str>) -> std::io::Error {
let mut bucket = "";
let mut object = "";
let mut version_id = "";
if !params.is_empty() {
bucket = params[0];
}
if params.len() >= 2 {
object = params[1];
}
if params.len() >= 3 {
version_id = params[2];
}
if is_network_or_host_down(&err.to_string(), false) {
return std::io::Error::other(ObjectApiError::BackendDown(format!("{err}")));
}
let err_ = std::io::Error::other(err.to_string());
let r_err = err;
let err;
let bucket = bucket.to_string();
let object = object.to_string();
let version_id = version_id.to_string();
match r_err.code {
S3ErrorCode::BucketNotEmpty => {
err = std::io::Error::other(StorageError::BucketNotEmpty("".to_string()).to_string());
}
S3ErrorCode::InvalidBucketName => {
err = std::io::Error::other(StorageError::BucketNameInvalid(bucket));
}
S3ErrorCode::InvalidPart => {
err = std::io::Error::other(StorageError::InvalidPart(0, bucket, object /* , version_id */));
}
S3ErrorCode::NoSuchBucket => {
err = std::io::Error::other(StorageError::BucketNotFound(bucket));
}
S3ErrorCode::NoSuchKey => {
if !object.is_empty() {
err = std::io::Error::other(StorageError::ObjectNotFound(bucket, object));
} else {
err = std::io::Error::other(StorageError::BucketNotFound(bucket));
}
}
S3ErrorCode::NoSuchVersion => {
if !object.is_empty() {
err = std::io::Error::other(StorageError::ObjectNotFound(bucket, object)); //, version_id);
} else {
err = std::io::Error::other(StorageError::BucketNotFound(bucket));
}
}
S3ErrorCode::AccessDenied => {
err = std::io::Error::other(StorageError::PrefixAccessDenied(bucket, object));
}
S3ErrorCode::NoSuchUpload => {
err = std::io::Error::other(StorageError::InvalidUploadID(bucket, object, version_id));
}
_ => {
err = err_;
}
}
err
}
#[cfg(test)]
mod tests {
use super::*;
-4
View File
@@ -20,10 +20,6 @@ use std::sync::atomic::AtomicI64;
/// this type never grew past its counter. `total_events` is read by the
/// notifier's log line but nothing increments it, so that field reports zero.
#[derive(Default)]
#[allow(
dead_code,
reason = "held only by the dead ecstore EventNotifier; see services/event_notification.rs (backlog#1823)"
)]
pub struct TargetList {
pub total_events: AtomicI64,
}
-1
View File
@@ -704,7 +704,6 @@ pub(crate) async fn create_bitrot_reader_from_bytes_with_stage_metrics(
}
#[allow(clippy::too_many_arguments)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn create_deferred_bitrot_reader(
inline_data: Option<Bytes>,
disk: Option<DiskStore>,
-22
View File
@@ -31,13 +31,6 @@ pub const ENV_DISK_COMPRESSION_MIME_TYPES: &str = "RUSTFS_COMPRESSION_MIME_TYPES
// Environment variable for additional extensions to exclude from compression (comma-separated, e.g. ".foo,.bar")
pub const ENV_ADDED_EXCLUDE_COMPRESS_EXTENSIONS: &str = "RUSTFS_ADDED_EXCLUDE_COMPRESS_EXTENSIONS";
// Environment variable to additionally enable disk compression for multipart uploads.
// Default off: nodes from before the resumable decompressor fix fail transient reads of
// compressed objects, so multipart compression stays dark until the operator confirms the
// fleet has converged on a fixed build.
// RUSTFS_COMPAT_TODO(multipart-compression-default-off-window): staged rollout switch for restored multipart compression, flipping the default to enabled on retirement. Remove after the minimum supported direct-upgrade release ships the resumable DecompressReader.
pub const ENV_DISK_COMPRESSION_MULTIPART_ENABLED: &str = "RUSTFS_COMPRESSION_MULTIPART_ENABLED";
pub const DEFAULT_DISK_COMPRESS_EXTENSIONS: &str = ".txt,.log,.csv,.json,.tar,.xml,.bin";
pub const DEFAULT_DISK_COMPRESS_MIME_TYPES: &str = "text/*,application/json,application/xml,binary/octet-stream";
@@ -178,21 +171,6 @@ pub fn is_disk_compression_enabled() -> bool {
DISK_COMPRESSION_CONFIG.get_or_init(parse_disk_compression_config).enabled
}
// Parsed once at first use, mirroring DISK_COMPRESSION_CONFIG.
static MULTIPART_DISK_COMPRESSION_ENABLED: OnceLock<bool> = OnceLock::new();
/// Whether multipart uploads may advertise disk compression. Requires the
/// regular disk-compression gates to pass as well; this is the staged-rollout
/// switch that keeps multipart compression dark during rolling upgrades from
/// builds whose decompressor was not yet resumable.
pub fn is_multipart_disk_compression_enabled() -> bool {
*MULTIPART_DISK_COMPRESSION_ENABLED.get_or_init(|| {
env::var(ENV_DISK_COMPRESSION_MULTIPART_ENABLED)
.map(|s| matches!(s.to_ascii_lowercase().as_str(), "true" | "on" | "1"))
.unwrap_or(false)
})
}
fn is_disk_compressible_with_config(headers: &http::HeaderMap, object_name: &str, config: &DiskCompressionConfig) -> bool {
// Check if disk compression is enabled (read once at first use, then fixed for process lifetime)
if !config.enabled {
+1 -1
View File
@@ -22,7 +22,7 @@ use crate::bucket::replication::{
use crate::bucket::versioning::VersioningApi as _;
use crate::config::storageclass;
use crate::error::{Error, Result};
use crate::io_support::rio::{HardLimitReader, HashReader};
use crate::io_support::rio::{HashReader, LimitReader};
use crate::storage_api_contracts::{
lifecycle::{ExpirationOptions, TransitionedObject},
range::HTTPRangeSpec,
+4 -475
View File
@@ -479,15 +479,7 @@ enum ReadTransform {
},
}
/// How an object's stored bytes must be fetched and transformed to serve a
/// request.
///
/// Public so callers that fetch the stored bytes from somewhere other than the
/// local erasure set — the remote-tier read path — can position their own fetch
/// with [`ReadPlan::storage_offset`] / [`ReadPlan::storage_length`] and then
/// hand the resulting stream to [`ReadPlan::into_object_reader`], instead of
/// reimplementing the transform decisions (rustfs/rustfs#6025).
pub struct ReadPlan {
struct ReadPlan {
storage_offset: usize,
storage_length: i64,
object_size: i64,
@@ -495,43 +487,6 @@ pub struct ReadPlan {
}
impl ReadPlan {
/// Byte offset into the object's **stored** bytes where the fetch must
/// start. Encrypted and compressed objects address their storage in a
/// different coordinate system than the plaintext range the caller asked
/// for, which is exactly the distinction this plan resolves.
pub fn storage_offset(&self) -> usize {
self.storage_offset
}
/// Number of **stored** bytes the fetch must deliver, in the same
/// coordinate system as [`Self::storage_offset`].
pub fn storage_length(&self) -> i64 {
self.storage_length
}
/// Build the plan for a request without consuming a stream, so a caller
/// that has to issue its own positioned fetch can read the offsets first.
pub async fn build_for_request(
rs: Option<HTTPRangeSpec>,
oi: &ObjectInfo,
opts: &ObjectOptions,
h: &HeaderMap<HeaderValue>,
resolver: Option<&dyn ObjectEncryptionResolver>,
) -> Result<Self> {
Self::build_with_resolver(rs, oi, opts, h, resolver).await
}
/// Wrap `reader` — the stored bytes this plan asked for, already positioned
/// at [`Self::storage_offset`] — in the transforms that turn them into the
/// bytes the caller requested.
pub fn into_object_reader(
self,
reader: Box<dyn AsyncRead + Unpin + Send + Sync>,
oi: &ObjectInfo,
) -> Result<GetObjectReader> {
self.into_reader(reader, oi).map(|(reader, _, _)| reader)
}
#[cfg(test)]
async fn build(rs: Option<HTTPRangeSpec>, oi: &ObjectInfo, opts: &ObjectOptions, h: &HeaderMap<HeaderValue>) -> Result<Self> {
Self::build_with_resolver(rs, oi, opts, h, Some(&tests::TEST_RESOLVER)).await
@@ -545,17 +500,8 @@ impl ReadPlan {
resolver: Option<&dyn ObjectEncryptionResolver>,
) -> Result<Self> {
let mut rs = rs;
// A part number addresses the object's PLAINTEXT bytes. A restore read
// serves the stored representation instead (see
// [`restore_request_active`]), where that synthesized range would be
// reinterpreted as a storage range and truncate an encrypted or
// compressed payload by exactly its encoding overhead — the copy-back
// then fails its length check partway through
// (rustfs/rustfs#6025). An explicit caller range is already in storage
// coordinates on that path and is still honored.
if let Some(part_number) = opts.part_number
&& rs.is_none()
&& !restore_request_active(opts)
{
rs = http_range_spec_from_object_info(oi, part_number);
}
@@ -808,7 +754,7 @@ impl ReadPlan {
}
}
} else {
Box::new(HardLimitReader::new(dec_reader, decompressed_length))
Box::new(LimitReader::new(dec_reader, total_plaintext_size))
};
let mut object_info = oi.clone();
@@ -900,7 +846,7 @@ impl ReadPlan {
)?;
Box::new(ranged_reader)
} else {
Box::new(HardLimitReader::new(decompressed_reader, total_plaintext_size_i64))
Box::new(LimitReader::new(decompressed_reader, total_plaintext_size))
}
} else if plaintext_offset > 0 || plaintext_length != total_plaintext_size_i64 {
Box::new(RangedDecompressReader::new(
@@ -910,7 +856,7 @@ impl ReadPlan {
total_plaintext_size,
)?)
} else {
Box::new(HardLimitReader::new(decrypted_reader, total_plaintext_size_i64))
Box::new(LimitReader::new(decrypted_reader, total_plaintext_size))
};
let mut object_info = oi.clone();
@@ -1781,423 +1727,6 @@ mod tests {
assert_eq!(actual, b"fghijkl");
}
/// Compresses one multipart part exactly like the write path does
/// (`WritePlan::with_compression` wraps each part in its own
/// `compression_reader`), returning the on-disk bytes and the storage-format
/// compression index.
async fn compressed_part_fixture(data: &[u8]) -> (Vec<u8>, Option<Bytes>) {
use crate::io_support::rio::TryGetIndex as _;
let mut compressor =
crate::io_support::rio::compression_reader(Cursor::new(data.to_vec()), CompressionAlgorithm::default(), false);
let mut compressed = Vec::new();
compressor.read_to_end(&mut compressed).await.expect("compress part stream");
let index = compressor
.try_get_index()
.map(crate::io_support::rio::compression_index_storage_bytes);
(compressed, index)
}
struct CompressedMultipartFixture {
object_info: ObjectInfo,
stored: Vec<u8>,
plaintext: Vec<u8>,
}
/// Builds the on-disk representation of a compressed multipart object: each
/// part is an independent compressed stream and the storage layer serves
/// their concatenation.
async fn compressed_multipart_fixture(part_sizes: &[usize]) -> CompressedMultipartFixture {
let pattern = b"compressed multipart read path fixture data ";
let mut plaintext = Vec::new();
let mut stored = Vec::new();
let mut parts = Vec::with_capacity(part_sizes.len());
for (i, part_size) in part_sizes.iter().enumerate() {
let mut part_plaintext = Vec::with_capacity(*part_size);
while part_plaintext.len() < *part_size {
part_plaintext.extend_from_slice(pattern);
part_plaintext.push(i as u8);
}
part_plaintext.truncate(*part_size);
let (compressed, index) = compressed_part_fixture(&part_plaintext).await;
parts.push(ObjectPartInfo {
number: i + 1,
size: compressed.len(),
actual_size: *part_size as i64,
index,
..Default::default()
});
stored.extend_from_slice(&compressed);
plaintext.extend_from_slice(&part_plaintext);
}
let mut user_defined = HashMap::new();
rustfs_utils::http::insert_str(
&mut user_defined,
rustfs_utils::http::SUFFIX_COMPRESSION,
crate::io_support::rio::compression_metadata_value(CompressionAlgorithm::default()),
);
rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, plaintext.len().to_string());
let object_info = ObjectInfo {
bucket: "test-bucket".to_string(),
name: "compressed-multipart".to_string(),
size: stored.len() as i64,
etag: Some(format!("6bcf86bed8807b8e78f0fc6e0a53079d-{}", part_sizes.len())),
parts: Arc::new(parts),
user_defined: Arc::new(user_defined),
..Default::default()
};
CompressedMultipartFixture {
object_info,
stored,
plaintext,
}
}
/// Plans the read once to learn the storage window, then serves exactly that
/// window — mirroring how `set_disk` feeds the erasure read into the
/// returned reader.
async fn read_compressed_multipart(
fixture: &CompressedMultipartFixture,
rs: Option<HTTPRangeSpec>,
opts: &ObjectOptions,
) -> Vec<u8> {
let headers = HeaderMap::new();
let (_, offset, length) =
GetObjectReader::new(Box::new(Cursor::new(Vec::new())), rs.clone(), &fixture.object_info, opts, &headers)
.await
.expect("plan compressed multipart read");
let end = offset + usize::try_from(length).expect("storage window length must be non-negative");
assert!(
end <= fixture.stored.len(),
"planned storage window {offset}..{end} exceeds stored stream of {} bytes",
fixture.stored.len()
);
let window = fixture.stored[offset..end].to_vec();
let (mut reader, replay_offset, replay_length) =
GetObjectReader::new(Box::new(Cursor::new(window)), rs, &fixture.object_info, opts, &headers)
.await
.expect("build compressed multipart reader");
assert_eq!((replay_offset, replay_length), (offset, length), "read plan must be deterministic");
reader.read_all().await.expect("read compressed multipart stream")
}
/// Byte pattern with a 2 KiB period: it compresses extremely well while
/// looking nothing like ASCII fixtures. Mirrors the e2e generator that
/// exposed a truncated full GET on high-ratio multipart payloads.
fn high_ratio_binary_payload(size: usize, seed: u8) -> Vec<u8> {
(0..size)
.map(|i| ((i as u64).wrapping_mul(2_654_435_761).wrapping_add(seed as u64) >> 3) as u8)
.collect()
}
#[tokio::test]
async fn compressed_multipart_full_get_handles_high_ratio_binary_payload() {
let part_sizes = [5 * 1024 * 1024_usize, 1024 * 1024];
let mut plaintext = Vec::new();
let mut stored = Vec::new();
let mut parts = Vec::with_capacity(part_sizes.len());
for (i, part_size) in part_sizes.iter().enumerate() {
let part_plaintext = high_ratio_binary_payload(*part_size, if i == 0 { 7 } else { 61 });
let (compressed, index) = compressed_part_fixture(&part_plaintext).await;
parts.push(ObjectPartInfo {
number: i + 1,
size: compressed.len(),
actual_size: *part_size as i64,
index,
..Default::default()
});
stored.extend_from_slice(&compressed);
plaintext.extend_from_slice(&part_plaintext);
}
let mut user_defined = HashMap::new();
rustfs_utils::http::insert_str(
&mut user_defined,
rustfs_utils::http::SUFFIX_COMPRESSION,
crate::io_support::rio::compression_metadata_value(CompressionAlgorithm::default()),
);
rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, plaintext.len().to_string());
let fixture = CompressedMultipartFixture {
object_info: ObjectInfo {
bucket: "test-bucket".to_string(),
name: "high-ratio-multipart".to_string(),
size: stored.len() as i64,
etag: Some("6bcf86bed8807b8e78f0fc6e0a53079d-2".to_string()),
parts: Arc::new(parts),
user_defined: Arc::new(user_defined),
..Default::default()
},
stored,
plaintext,
};
let read = read_compressed_multipart(&fixture, None, &ObjectOptions::default()).await;
assert_eq!(read.len(), fixture.plaintext.len(), "full GET must return the logical size");
assert_eq!(read, fixture.plaintext, "high-ratio multipart payload must survive the roundtrip");
}
/// Full GET over a compressed multipart object must decode across part
/// boundaries: every part is an independent compressed stream (this is also
/// the on-disk shape written by builds before rustfs/rustfs#5169 disabled
/// multipart compression, so this pins legacy-object readability).
#[tokio::test]
async fn compressed_multipart_full_get_decodes_across_part_boundaries() {
let fixture = compressed_multipart_fixture(&[3 * 1024 * 1024, 2 * 1024 * 1024, 512 * 1024]).await;
let read = read_compressed_multipart(&fixture, None, &ObjectOptions::default()).await;
assert_eq!(read.len(), fixture.plaintext.len(), "full GET must return the logical size");
assert_eq!(read, fixture.plaintext, "full GET must reassemble all parts");
}
#[tokio::test]
async fn compressed_multipart_range_get_crosses_part_boundary() {
let fixture = compressed_multipart_fixture(&[3 * 1024 * 1024, 2 * 1024 * 1024]).await;
let boundary = 3 * 1024 * 1024_i64;
let rs = HTTPRangeSpec {
is_suffix_length: false,
start: boundary - 100_000,
end: boundary + 100_000 - 1,
};
let read = read_compressed_multipart(&fixture, Some(rs), &ObjectOptions::default()).await;
let expected = &fixture.plaintext[(boundary - 100_000) as usize..(boundary + 100_000) as usize];
assert_eq!(read, expected, "boundary-crossing range must splice both parts");
}
#[tokio::test]
async fn compressed_multipart_range_get_seeks_into_later_part() {
let fixture = compressed_multipart_fixture(&[3 * 1024 * 1024, 4 * 1024 * 1024]).await;
// Deep inside part 2 so the plan skips part 1 entirely and (when the
// part carries an index) seeks within part 2.
let start = 3 * 1024 * 1024_i64 + 2 * 1024 * 1024_i64 + 137;
let rs = HTTPRangeSpec {
is_suffix_length: false,
start,
end: start + 64 * 1024 - 1,
};
let read = read_compressed_multipart(&fixture, Some(rs), &ObjectOptions::default()).await;
let expected = &fixture.plaintext[start as usize..(start + 64 * 1024) as usize];
assert_eq!(read, expected, "range inside a later part must decode from that part");
}
/// Parts written without a compression index (small parts skip the index in
/// the rio-v2 backend) must still be rangeable: the plan starts at the part
/// boundary and skips decompressed bytes.
#[tokio::test]
async fn compressed_multipart_range_get_works_without_part_indexes() {
let mut fixture = compressed_multipart_fixture(&[1024 * 1024, 1024 * 1024]).await;
let parts = fixture
.object_info
.parts
.iter()
.map(|part| ObjectPartInfo {
index: None,
..part.clone()
})
.collect::<Vec<_>>();
fixture.object_info.parts = Arc::new(parts);
let start = 1024 * 1024_i64 + 4096;
let rs = HTTPRangeSpec {
is_suffix_length: false,
start,
end: start + 32 * 1024 - 1,
};
let read = read_compressed_multipart(&fixture, Some(rs), &ObjectOptions::default()).await;
let expected = &fixture.plaintext[start as usize..(start + 32 * 1024) as usize];
assert_eq!(read, expected, "index-less parts must fall back to part-boundary skip");
}
#[tokio::test]
async fn compressed_multipart_part_number_get_returns_single_part() {
let part_sizes = [3 * 1024 * 1024, 2 * 1024 * 1024, 512 * 1024];
let fixture = compressed_multipart_fixture(&part_sizes).await;
let mut logical_offset = 0_usize;
for (i, part_size) in part_sizes.iter().enumerate() {
let opts = ObjectOptions {
part_number: Some(i + 1),
..Default::default()
};
let read = read_compressed_multipart(&fixture, None, &opts).await;
let expected = &fixture.plaintext[logical_offset..logical_offset + part_size];
assert_eq!(read.len(), *part_size, "partNumber={} GET must return the part's logical size", i + 1);
assert_eq!(read, expected, "partNumber={} GET must return the original part bytes", i + 1);
logical_offset += part_size;
}
}
#[tokio::test]
async fn compressed_multipart_suffix_range_reads_tail() {
let fixture = compressed_multipart_fixture(&[3 * 1024 * 1024, 1024 * 1024]).await;
let suffix_len = 128 * 1024_i64;
let rs = HTTPRangeSpec {
is_suffix_length: true,
start: suffix_len,
end: -1,
};
let read = read_compressed_multipart(&fixture, Some(rs), &ObjectOptions::default()).await;
let expected = &fixture.plaintext[fixture.plaintext.len() - suffix_len as usize..];
assert_eq!(read, expected, "suffix range must return the tail of the last part");
}
/// Builds an SSE-C + disk-compression multipart object exactly like the
/// write path: each part is compressed into its own stream and then
/// encrypted with the per-part key schedule. The fixture is
/// legacy-encryption-specific (`rustfs_rio::EncryptReader`), matching the
/// pre-existing `build_legacy_ssec_multipart_fixture` shape, while the
/// compression layer follows the active backend feature.
async fn compressed_encrypted_multipart_fixture(key_bytes: [u8; 32], part_sizes: &[usize]) -> CompressedMultipartFixture {
let pattern = b"compressed encrypted multipart fixture data ";
let mut plaintext = Vec::new();
let mut stored = Vec::new();
let mut parts = Vec::with_capacity(part_sizes.len());
for (i, part_size) in part_sizes.iter().enumerate() {
let part_number = i + 1;
let mut part_plaintext = Vec::with_capacity(*part_size);
while part_plaintext.len() < *part_size {
part_plaintext.extend_from_slice(pattern);
part_plaintext.push(part_number as u8);
}
part_plaintext.truncate(*part_size);
let (compressed, index) = compressed_part_fixture(&part_plaintext).await;
let mut part_cipher = Vec::new();
rustfs_rio::EncryptReader::new_multipart(Cursor::new(compressed), key_bytes, LEGACY_FIXTURE_BASE_NONCE, part_number)
.read_to_end(&mut part_cipher)
.await
.expect("encrypt compressed fixture part");
parts.push(ObjectPartInfo {
number: part_number,
size: part_cipher.len(),
actual_size: *part_size as i64,
index,
..Default::default()
});
stored.extend_from_slice(&part_cipher);
plaintext.extend_from_slice(&part_plaintext);
}
let mut user_defined = legacy_ssec_multipart_metadata(key_bytes, plaintext.len());
rustfs_utils::http::insert_str(
&mut user_defined,
rustfs_utils::http::SUFFIX_COMPRESSION,
crate::io_support::rio::compression_metadata_value(CompressionAlgorithm::default()),
);
rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, plaintext.len().to_string());
let object_info = ObjectInfo {
bucket: "test-bucket".to_string(),
name: "compressed-encrypted-multipart".to_string(),
size: stored.len() as i64,
etag: Some(format!("6bcf86bed8807b8e78f0fc6e0a53079d-{}", part_sizes.len())),
parts: Arc::new(parts),
user_defined: Arc::new(user_defined),
..Default::default()
};
CompressedMultipartFixture {
object_info,
stored,
plaintext,
}
}
async fn read_compressed_encrypted_multipart(
fixture: &CompressedMultipartFixture,
key_bytes: [u8; 32],
rs: Option<HTTPRangeSpec>,
opts: &ObjectOptions,
) -> Vec<u8> {
let headers = ssec_headers_from_key(key_bytes);
let (_, offset, length) =
GetObjectReader::new(Box::new(Cursor::new(Vec::new())), rs.clone(), &fixture.object_info, opts, &headers)
.await
.expect("plan compressed encrypted multipart read");
let end = offset + usize::try_from(length).expect("storage window length must be non-negative");
assert!(
end <= fixture.stored.len(),
"planned storage window {offset}..{end} exceeds stored stream of {} bytes",
fixture.stored.len()
);
let window = fixture.stored[offset..end].to_vec();
let (mut reader, replay_offset, replay_length) =
GetObjectReader::new(Box::new(Cursor::new(window)), rs, &fixture.object_info, opts, &headers)
.await
.expect("build compressed encrypted multipart reader");
assert_eq!((replay_offset, replay_length), (offset, length), "read plan must be deterministic");
reader.read_all().await.expect("read compressed encrypted multipart stream")
}
#[tokio::test]
async fn compressed_encrypted_multipart_full_get_roundtrip() {
let key_bytes = [0x6Eu8; 32];
let fixture = compressed_encrypted_multipart_fixture(key_bytes, &[3 * 1024 * 1024, 1024 * 1024]).await;
let read = read_compressed_encrypted_multipart(&fixture, key_bytes, None, &ObjectOptions::default()).await;
assert_eq!(read.len(), fixture.plaintext.len(), "full GET must return the logical size");
assert_eq!(read, fixture.plaintext, "SSE-C + compression full GET must reassemble all parts");
}
#[tokio::test]
async fn compressed_encrypted_multipart_range_crosses_part_boundary() {
let key_bytes = [0x6Eu8; 32];
let fixture = compressed_encrypted_multipart_fixture(key_bytes, &[3 * 1024 * 1024, 1024 * 1024]).await;
let boundary = 3 * 1024 * 1024_i64;
let rs = HTTPRangeSpec {
is_suffix_length: false,
start: boundary - 65_536,
end: boundary + 65_536 - 1,
};
let read = read_compressed_encrypted_multipart(&fixture, key_bytes, Some(rs), &ObjectOptions::default()).await;
let expected = &fixture.plaintext[(boundary - 65_536) as usize..(boundary + 65_536) as usize];
assert_eq!(read, expected, "SSE-C + compression boundary-crossing range must splice both parts");
}
#[tokio::test]
async fn compressed_encrypted_multipart_part_number_get_returns_single_part() {
let key_bytes = [0x6Eu8; 32];
let part_sizes = [3 * 1024 * 1024, 1024 * 1024];
let fixture = compressed_encrypted_multipart_fixture(key_bytes, &part_sizes).await;
let opts = ObjectOptions {
part_number: Some(2),
..Default::default()
};
let read = read_compressed_encrypted_multipart(&fixture, key_bytes, None, &opts).await;
let expected = &fixture.plaintext[part_sizes[0]..];
assert_eq!(read.len(), part_sizes[1], "partNumber=2 GET must return the part's logical size");
assert_eq!(read, expected, "partNumber=2 GET must return the original part bytes");
}
#[tokio::test]
async fn test_get_object_reader_rejects_ssec_read_without_headers() {
let object_info = ObjectInfo {
-6
View File
@@ -277,12 +277,6 @@ pub struct ObjectOptions {
/// fence avoids recursively acquiring the read lock behind a queued writer.
pub bucket_lifecycle_lock_fence: Option<NamespaceLockFence>,
pub replication_request: bool,
/// Source-cluster LWW timestamps carried by an authorized replication
/// request; None when the source never modified the category. Only the
/// replication-authorized options builders may set these.
pub replication_tagging_timestamp: Option<OffsetDateTime>,
pub replication_retention_timestamp: Option<OffsetDateTime>,
pub replication_legalhold_timestamp: Option<OffsetDateTime>,
/// Authorized SSE-C replication passthrough: the body is already
/// ciphertext, so the write path must not encrypt or compress it and
/// stores the restored encryption metadata verbatim. Only the
@@ -23,10 +23,6 @@ use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::task::JoinSet;
#[allow(
dead_code,
reason = "default operation label for the test-only AsyncBatchProcessor::new (backlog#1823)"
)]
const BATCH_PROCESSOR_OPERATION_CUSTOM: &str = "custom";
const BATCH_PROCESSOR_OPERATION_READ: &str = "read";
const BATCH_PROCESSOR_OPERATION_WRITE: &str = "write";
@@ -215,7 +211,6 @@ pub struct AsyncBatchProcessor {
}
impl AsyncBatchProcessor {
#[allow(dead_code, reason = "constructor used only by this file's tests (backlog#1823)")]
pub fn new(max_concurrent: usize) -> Self {
Self::new_with_operation(max_concurrent, BATCH_PROCESSOR_OPERATION_CUSTOM)
}
@@ -26,26 +26,11 @@ use std::sync::atomic::Ordering;
use tokio::sync::RwLock;
use tracing::warn;
/// Dead ecstore-side notification skeleton.
///
/// The working notification stack is `rustfs-notify`, whose own `EventNotifier`
/// is the one bucket configuration actually drives. Nothing calls the methods
/// below; `init_bucket_targets` even logs that it is a no-op in this build.
/// Removing it means also retiring the `InstanceContext` slot that holds it
/// (backlog#939 Phase 5), so it is left explicit here rather than half-removed.
#[allow(
dead_code,
reason = "ecstore-side notification skeleton superseded by rustfs-notify; see module note (backlog#1823)"
)]
pub struct EventNotifier {
target_list: TargetList,
//bucket_rules_map: HashMap<String , HashMap<EventName, Rules>>,
}
#[allow(
dead_code,
reason = "ecstore-side notification skeleton superseded by rustfs-notify; see module note (backlog#1823)"
)]
impl EventNotifier {
pub fn new() -> Arc<RwLock<Self>> {
Arc::new(RwLock::new(Self {
+1
View File
@@ -13,6 +13,7 @@
// limitations under the License.
// #730: background service owners still contain staged notification/rebalance/tier paths.
#![allow(dead_code)]
pub(crate) mod batch_processor;
pub(crate) mod event_notification;
@@ -1623,7 +1623,6 @@ impl NotificationSys {
workers.peers.remove(host);
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn tier_config_reload_worker_active(&self, host: &str) -> bool {
self.tier_config_reload_workers
.lock()
@@ -1797,7 +1796,6 @@ where
.map_err(|_| Error::other(format!("scanner activity peer {host} timed out after {timeout_duration:?}")))?
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
async fn call_peer_with_timeout<F, Fut>(
timeout_dur: Duration,
host_label: &str,
@@ -864,10 +864,6 @@ pub(super) fn merge_rebalance_meta(remote: &mut RebalanceMeta, local: &Rebalance
RebalanceMetaMergeOutcome::Merged
}
#[allow(
dead_code,
reason = "stop-transition helper retained beside stop_rebalance_meta_snapshot; no caller yet (backlog#1823)"
)]
pub(super) fn mark_started_rebalance_pools_stopped(meta: &mut RebalanceMeta, stop_time: OffsetDateTime) {
for pool_stat in meta.pool_stats.iter_mut() {
if pool_stat.info.status == RebalStatus::Started {
@@ -968,7 +964,6 @@ pub(super) fn rollback_rebalance_start_meta_snapshot_for_id(
})
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(super) fn stop_rebalance_meta_snapshot(meta: Option<&mut RebalanceMeta>, now: OffsetDateTime) -> Option<RebalanceMeta> {
let meta = meta?;
stop_rebalance_state(meta, now);
@@ -171,7 +171,6 @@ where
}
#[allow(clippy::too_many_arguments)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(super) async fn migrate_entry_version_with_retry_wait<Backend, F, Fut, D, DFut, W, WFut>(
set: &Backend,
bucket: String,
@@ -1,4 +1,5 @@
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use time::OffsetDateTime;
use tokio_util::sync::CancellationToken;
@@ -31,6 +32,8 @@ pub struct RebalanceStats {
pub cleanup_warnings: RebalanceCleanupWarnings,
}
pub type RStats = Vec<Arc<RebalanceStats>>;
#[derive(Debug, Default)]
pub(super) struct RebalanceBucketConfigs {
pub(super) bucket_incarnation_id: Option<uuid::Uuid>,
@@ -132,6 +135,7 @@ impl RebalanceStopPropagationRecord {
}
}
#[allow(dead_code)]
#[derive(Debug, Clone, Default)]
pub struct DiskStat {
pub total_space: u64,
+1
View File
@@ -30,5 +30,6 @@ pub mod warm_backend_minio;
pub mod warm_backend_r2;
pub mod warm_backend_rustfs;
pub mod warm_backend_s3;
pub mod warm_backend_s3sdk;
pub mod warm_backend_tencent;
pub mod warm_backend_wasabi;
+20 -10
View File
@@ -488,7 +488,6 @@ impl TierCandidateMutation {
targets
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn affected_targets(
&self,
manager: &TierConfigMgr,
@@ -803,7 +802,6 @@ fn tier_persisted_reference_blocks_any_target(
.any(|target| tier_persisted_reference_blocks_target(tier_name, backend_identity, target))
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn tier_object_blocks_target_rebind(object: &ObjectInfo, target: &TierMutationIntentTarget) -> io::Result<bool> {
tier_object_blocks_any_target_rebind(object, std::slice::from_ref(target))
}
@@ -2728,6 +2726,14 @@ impl TierConfigMgr {
Self::publish_candidate_owned(handle, candidate, driver_tier.map(str::to_string), update).await
}
fn begin_publish_transition(
handle: &Arc<RwLock<Self>>,
manager: &mut Self,
candidate: &Self,
) -> std::result::Result<TierPublishTransition, AdminError> {
Self::begin_publish_transition_with_allowed_mutation_blocks(handle, manager, candidate, None)
}
fn begin_publish_transition_with_allowed_mutation_blocks(
handle: &Arc<RwLock<Self>>,
manager: &mut Self,
@@ -2813,6 +2819,14 @@ impl TierConfigMgr {
})
}
async fn publish_candidate_inner(
handle: &Arc<RwLock<Self>>,
candidate: Self,
driver_tier: Option<&str>,
) -> std::result::Result<(), AdminError> {
Self::publish_candidate_inner_with_allowed_mutation_blocks(handle, candidate, driver_tier, None).await
}
async fn publish_candidate_inner_with_allowed_mutation_blocks(
handle: &Arc<RwLock<Self>>,
candidate: Self,
@@ -2925,7 +2939,6 @@ impl TierConfigMgr {
admin_err
}
#[allow(dead_code, reason = "reached only through #[cfg(test)] helpers in this file (backlog#1823)")]
async fn publish_candidate_owned(
handle: &Arc<RwLock<Self>>,
candidate: Self,
@@ -3528,7 +3541,6 @@ impl TierConfigMgr {
Self::update_candidate_with_config_lock(handle, api, TierCandidateMutation::Remove(tier_name.to_string(), force)).await
}
#[allow(dead_code, reason = "reached only through #[cfg(test)] helpers in this file (backlog#1823)")]
async fn remove_and_save_with<S>(
handle: &Arc<RwLock<Self>>,
api: Arc<S>,
@@ -3562,7 +3574,6 @@ impl TierConfigMgr {
Self::update_candidate_with_config_lock(handle, api, TierCandidateMutation::Clear(force)).await
}
#[allow(dead_code, reason = "reached only through #[cfg(test)] helpers in this file (backlog#1823)")]
async fn clear_and_save_with<S>(
handle: &Arc<RwLock<Self>>,
api: Arc<S>,
@@ -3601,10 +3612,6 @@ impl TierConfigMgr {
}
#[cfg(test)]
#[allow(
dead_code,
reason = "lease accounting asserted by a bucket_lifecycle_ops test behind `--features test-util` (backlog#1823)"
)]
pub(crate) async fn active_operation_lease_count(handle: &Arc<RwLock<Self>>, tier_name: &str) -> usize {
let manager = handle.read().await;
let Some(runtime) = registered_tier_driver_runtime(&manager) else {
@@ -3710,6 +3717,10 @@ impl TierConfigMgr {
Ok(())
}
fn retire_driver(&mut self, tier_name: &str) {
self.revoke_driver(tier_name);
}
fn revoke_all_drivers(&mut self) {
if let Some(runtime) = registered_tier_driver_runtime(self) {
let mut runtime = lock_unpoisoned(&runtime);
@@ -3873,7 +3884,6 @@ impl TierConfigMgr {
self.save_config(api, &config_file, data).await
}
#[allow(dead_code, reason = "reached only through #[cfg(test)] helpers in this file (backlog#1823)")]
async fn save_tiering_config_if_current<S>(
&self,
api: Arc<S>,
@@ -16,16 +16,8 @@ use serde::{Deserialize, Serialize};
use std::{fmt::Display, io};
use tracing::info;
#[allow(
dead_code,
reason = "tier config wire version stamped by the parity constructors below (backlog#1823)"
)]
const C_TIER_CONFIG_VER: &str = "v1";
#[allow(
dead_code,
reason = "tier-name validation message reached only from the parity constructors below (backlog#1823)"
)]
const ERR_TIER_NAME_EMPTY: &str = "remote tier name empty";
const WASABI_US_EAST_ENDPOINT: &str = "https://s3.wasabisys.com";
const WASABI_ALTERNATIVE_ENDPOINTS: &[(&str, &str)] = &[
@@ -272,6 +264,7 @@ impl Clone for TierConfig {
}
}
#[allow(dead_code)]
impl TierConfig {
pub(crate) fn clone_with_credentials(&self) -> Self {
Self {
@@ -291,7 +284,6 @@ impl TierConfig {
}
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
fn endpoint(&self) -> String {
match self.tier_type {
TierType::S3 => self.s3.as_ref().map(|s| s.endpoint.clone()).unwrap_or_default(),
@@ -311,7 +303,6 @@ impl TierConfig {
}
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
fn bucket(&self) -> String {
match self.tier_type {
TierType::S3 => self.s3.as_ref().map(|s| s.bucket.clone()).unwrap_or_default(),
@@ -331,7 +322,6 @@ impl TierConfig {
}
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
fn prefix(&self) -> String {
match self.tier_type {
TierType::S3 => self.s3.as_ref().map(|s| s.prefix.clone()).unwrap_or_default(),
@@ -351,7 +341,6 @@ impl TierConfig {
}
}
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
fn region(&self) -> String {
match self.tier_type {
TierType::S3 => self.s3.as_ref().map(|s| s.region.clone()).unwrap_or_default(),
@@ -468,7 +457,7 @@ impl TierWasabi {
}
impl TierS3 {
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
#[allow(dead_code)]
fn create<F>(
name: &str,
access_key: &str,
@@ -539,7 +528,7 @@ pub struct TierMinIO {
}
impl TierMinIO {
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
#[allow(dead_code)]
fn create<F>(
name: &str,
endpoint: &str,
@@ -14,6 +14,7 @@
use crate::services::tier::tier::TierConfigMgr;
#[allow(dead_code)]
impl TierConfigMgr {
pub fn msg_size(&self) -> usize {
100

Some files were not shown because too many files have changed in this diff Show More