mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-31 18:42:17 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 overtrue
|
aa1a8c1b3f | ||
|
Zhengchao An
|
f718e72e24 | ||
|
houseme
|
e06c9c02c6 |
Generated
+45
-32
@@ -1528,7 +1528,7 @@ version = "0.10.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71"
|
||||
dependencies = [
|
||||
"generic-array 0.14.9",
|
||||
"generic-array 0.14.7",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -1547,7 +1547,7 @@ version = "0.3.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a8894febbff9f758034a5b8e12d87918f56dfc64a8e1fe757d65e29041538d93"
|
||||
dependencies = [
|
||||
"generic-array 0.14.9",
|
||||
"generic-array 0.14.7",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -1598,7 +1598,7 @@ version = "3.9.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6dee98b0db6a962de883bf5d20362dee4d7ca0d12fe39a7c6c73c844e1cd7c1f"
|
||||
dependencies = [
|
||||
"darling 0.23.0",
|
||||
"darling 0.20.11",
|
||||
"ident_case",
|
||||
"prettyplease",
|
||||
"proc-macro2",
|
||||
@@ -1870,7 +1870,7 @@ version = "0.4.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad"
|
||||
dependencies = [
|
||||
"crypto-common 0.1.6",
|
||||
"crypto-common 0.1.7",
|
||||
"inout 0.1.4",
|
||||
]
|
||||
|
||||
@@ -2328,7 +2328,7 @@ version = "0.5.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0dc92fb57ca44df6db8059111ab3af99a63d5d0f8375d9972e319a379c6bab76"
|
||||
dependencies = [
|
||||
"generic-array 0.14.9",
|
||||
"generic-array 0.14.7",
|
||||
"rand_core 0.6.4",
|
||||
"subtle",
|
||||
"zeroize",
|
||||
@@ -2353,11 +2353,11 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "crypto-common"
|
||||
version = "0.1.6"
|
||||
version = "0.1.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3"
|
||||
checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a"
|
||||
dependencies = [
|
||||
"generic-array 0.14.9",
|
||||
"generic-array 0.14.7",
|
||||
"typenum",
|
||||
]
|
||||
|
||||
@@ -3554,7 +3554,7 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
|
||||
dependencies = [
|
||||
"block-buffer 0.10.4",
|
||||
"const-oid 0.9.6",
|
||||
"crypto-common 0.1.6",
|
||||
"crypto-common 0.1.7",
|
||||
"subtle",
|
||||
]
|
||||
|
||||
@@ -3813,7 +3813,7 @@ dependencies = [
|
||||
"crypto-bigint 0.5.5",
|
||||
"digest 0.10.7",
|
||||
"ff 0.13.1",
|
||||
"generic-array 0.14.9",
|
||||
"generic-array 0.14.7",
|
||||
"group 0.13.0",
|
||||
"hkdf 0.12.4",
|
||||
"pem-rfc7468 0.7.0",
|
||||
@@ -4258,9 +4258,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "generic-array"
|
||||
version = "0.14.9"
|
||||
version = "0.14.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4bb6743198531e02858aeaea5398fcc883e71851fcbcb5a2f773e2fb6cb1edf2"
|
||||
checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"
|
||||
dependencies = [
|
||||
"typenum",
|
||||
"version_check",
|
||||
@@ -4273,7 +4273,7 @@ version = "1.4.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ab4e5aa225bc56696909483320f0ff9b600f1a971b52e07a17d70f3d9b43254b"
|
||||
dependencies = [
|
||||
"generic-array 0.14.9",
|
||||
"generic-array 0.14.7",
|
||||
"rustversion",
|
||||
"typenum",
|
||||
]
|
||||
@@ -4372,9 +4372,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "google-cloud-auth"
|
||||
version = "1.14.0"
|
||||
version = "1.15.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a3494870d06f3cbbb3561ada6f234982549e3a2fb31e719ef258e6eadb9ae09a"
|
||||
checksum = "f54aab44c16b8463ae11b165a87c3d484780231f157bb1ed65843d591beb5abd"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"aws-lc-rs",
|
||||
@@ -4401,9 +4401,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "google-cloud-gax"
|
||||
version = "1.12.0"
|
||||
version = "1.13.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3103a4a9013f1aed573ca56e19a9680b0211643a99ea85caf524b397d6be8be3"
|
||||
checksum = "b9a46dd0fd026bbc4a5d84e6ab0c941cee6e3b057976a0bb107fdb5238ce598f"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures",
|
||||
@@ -4420,9 +4420,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "google-cloud-gax-internal"
|
||||
version = "0.7.15"
|
||||
version = "0.7.16"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c0df265fba091ed7e00ecd0755009423310163f8b52820f007b6b4d97f4c6617"
|
||||
checksum = "fb04c54317ace06d489213f761797240b3046142a9b7ce6b9a82a9d134e193d1"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures",
|
||||
@@ -4524,9 +4524,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "google-cloud-storage"
|
||||
version = "1.16.0"
|
||||
version = "1.17.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "dc4b1d78c88db5c2530b12461e373a7d0d3a6caa3f6c1fc14e5d824cf2aeb307"
|
||||
checksum = "9227f65175fa91a6e41f246797917697efdadfe09dd8ea84ad8b737a71efbd28"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"base64 0.22.1",
|
||||
@@ -4577,9 +4577,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "google-cloud-wkt"
|
||||
version = "1.6.0"
|
||||
version = "1.7.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "46df1fcc3ab69164af3f4199ed21f45b5dbc56d9f03211eb4fa20116d442364b"
|
||||
checksum = "7fccf98cfd5481a5f5a285181ab0c62123d7d47cd2bb7299448440649349e4e7"
|
||||
dependencies = [
|
||||
"base64 0.22.1",
|
||||
"bytes",
|
||||
@@ -4920,6 +4920,7 @@ dependencies = [
|
||||
"async-trait",
|
||||
"cfg-if",
|
||||
"crossbeam-channel",
|
||||
"flate2",
|
||||
"futures-channel",
|
||||
"futures-util",
|
||||
"hdrhistogram",
|
||||
@@ -4927,6 +4928,7 @@ dependencies = [
|
||||
"hotpath-meta",
|
||||
"http 1.5.0",
|
||||
"libc",
|
||||
"object 0.36.7",
|
||||
"parking_lot",
|
||||
"pin-project-lite",
|
||||
"prettytable-rs",
|
||||
@@ -4934,6 +4936,7 @@ dependencies = [
|
||||
"regex",
|
||||
"reqwest",
|
||||
"reqwest-middleware",
|
||||
"rustc-demangle",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tiny_http",
|
||||
@@ -5047,9 +5050,9 @@ checksum = "15cdd26707701c53297e2fa6afb323d55fbc1d0810c3aec078ae3ef0424c3c15"
|
||||
|
||||
[[package]]
|
||||
name = "hybrid-array"
|
||||
version = "0.4.13"
|
||||
version = "0.4.14"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "818356c5132c1fede50f837ca96afbe78ff42413047f4abb886217845e1b6c8c"
|
||||
checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b"
|
||||
dependencies = [
|
||||
"ctutils",
|
||||
"subtle",
|
||||
@@ -5297,7 +5300,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01"
|
||||
dependencies = [
|
||||
"block-padding 0.3.3",
|
||||
"generic-array 0.14.9",
|
||||
"generic-array 0.14.7",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -6807,6 +6810,15 @@ dependencies = [
|
||||
"objc2-core-foundation",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "object"
|
||||
version = "0.36.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "62948e14d923ea95ea2c7c86c71013138b66525b86bdc08d2dcc262bdb497b87"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "object"
|
||||
version = "0.37.3"
|
||||
@@ -7844,7 +7856,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf"
|
||||
dependencies = [
|
||||
"heck",
|
||||
"itertools 0.10.5",
|
||||
"itertools 0.13.0",
|
||||
"log",
|
||||
"multimap",
|
||||
"once_cell",
|
||||
@@ -7864,7 +7876,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042"
|
||||
dependencies = [
|
||||
"heck",
|
||||
"itertools 0.10.5",
|
||||
"itertools 0.13.0",
|
||||
"log",
|
||||
"multimap",
|
||||
"petgraph 0.8.3",
|
||||
@@ -7885,7 +7897,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"itertools 0.10.5",
|
||||
"itertools 0.13.0",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.119",
|
||||
@@ -7898,7 +7910,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"itertools 0.10.5",
|
||||
"itertools 0.13.0",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.119",
|
||||
@@ -9716,6 +9728,7 @@ dependencies = [
|
||||
"base64-simd",
|
||||
"chrono",
|
||||
"futures",
|
||||
"hotpath",
|
||||
"ipnetwork",
|
||||
"jsonwebtoken 11.0.0",
|
||||
"moka",
|
||||
@@ -10550,7 +10563,7 @@ checksum = "d3e97a565f76233a6003f9f5c54be1d9c5bdfa3eccfb189469f11ec4901c47dc"
|
||||
dependencies = [
|
||||
"base16ct 0.2.0",
|
||||
"der 0.7.10",
|
||||
"generic-array 0.14.9",
|
||||
"generic-array 0.14.7",
|
||||
"pkcs8 0.10.2",
|
||||
"subtle",
|
||||
"zeroize",
|
||||
@@ -11540,7 +11553,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
|
||||
dependencies = [
|
||||
"fastrand",
|
||||
"getrandom 0.3.4",
|
||||
"getrandom 0.4.3",
|
||||
"once_cell",
|
||||
"rustix",
|
||||
"windows-sys 0.61.2",
|
||||
|
||||
+2
-2
@@ -250,8 +250,8 @@ enumset = "1.1.14"
|
||||
faster-hex = "0.10.0"
|
||||
flate2 = "1.1.9"
|
||||
glob = "0.3.4"
|
||||
google-cloud-storage = "1.16.0"
|
||||
google-cloud-auth = "1.14.0"
|
||||
google-cloud-storage = "1.17.0"
|
||||
google-cloud-auth = "1.15.0"
|
||||
hashbrown = { version = "0.17.1" }
|
||||
hex = "0.4.3"
|
||||
hex-simd = "0.8.0"
|
||||
|
||||
@@ -26,7 +26,6 @@
|
||||
|
||||
use super::common::*;
|
||||
use aws_sdk_s3::Client;
|
||||
use aws_sdk_s3::error::ProvideErrorMetadata;
|
||||
use aws_sdk_s3::primitives::{ByteStream, DateTimeFormat};
|
||||
use aws_sdk_s3::types::{
|
||||
CompletedMultipartUpload, CompletedPart, Delete, MetadataDirective, ObjectIdentifier, ObjectLockLegalHoldStatus,
|
||||
@@ -2121,127 +2120,6 @@ async fn test_multipart_default_retention_fixed_at_create() {
|
||||
// Versioning Auto-Enable Tests
|
||||
// ============================================================================
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_unretained_object_lock_object_delete_and_bucket_cleanup() {
|
||||
init_logging();
|
||||
info!("🧪 Test: Unretained Object Lock object delete and bucket cleanup (Issue #5339)");
|
||||
|
||||
let mut env = ObjectLockTestEnvironment::new()
|
||||
.await
|
||||
.expect("failed to create Object Lock test environment");
|
||||
env.start_rustfs().await.expect("failed to start RustFS");
|
||||
|
||||
let bucket = "test-object-lock-delete-cleanup";
|
||||
let key = "unretained-object";
|
||||
|
||||
env.create_object_lock_bucket(bucket)
|
||||
.await
|
||||
.expect("failed to create Object Lock bucket");
|
||||
let client = env.s3_client();
|
||||
|
||||
let put_response = client
|
||||
.put_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.body(ByteStream::from_static(b"unretained data"))
|
||||
.send()
|
||||
.await
|
||||
.expect("failed to upload unretained object");
|
||||
let object_version_id = put_response
|
||||
.version_id()
|
||||
.expect("Object Lock buckets must create versioned objects")
|
||||
.to_string();
|
||||
|
||||
let delete_response = client
|
||||
.delete_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.send()
|
||||
.await
|
||||
.expect("failed to create delete marker");
|
||||
assert_eq!(delete_response.delete_marker(), Some(true));
|
||||
let delete_marker_version_id = delete_response
|
||||
.version_id()
|
||||
.expect("Deleting without a version ID must create a delete marker")
|
||||
.to_string();
|
||||
|
||||
let get_error = client
|
||||
.get_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.send()
|
||||
.await
|
||||
.expect_err("GET must not return an object hidden by a delete marker");
|
||||
assert_eq!(get_error.raw_response().map(|response| response.status().as_u16()), Some(404));
|
||||
assert_eq!(get_error.as_service_error().and_then(|error| error.code()), Some("NoSuchKey"));
|
||||
|
||||
let listed_objects = client
|
||||
.list_objects_v2()
|
||||
.bucket(bucket)
|
||||
.send()
|
||||
.await
|
||||
.expect("failed to list current objects");
|
||||
assert!(
|
||||
listed_objects.contents().iter().all(|object| object.key() != Some(key)),
|
||||
"ListObjectsV2 must hide objects whose latest version is a delete marker"
|
||||
);
|
||||
|
||||
let listed_versions = client
|
||||
.list_object_versions()
|
||||
.bucket(bucket)
|
||||
.send()
|
||||
.await
|
||||
.expect("failed to list object versions");
|
||||
assert!(
|
||||
listed_versions
|
||||
.versions()
|
||||
.iter()
|
||||
.any(|version| version.key() == Some(key) && version.version_id() == Some(object_version_id.as_str())),
|
||||
"The data version must remain until it is explicitly deleted"
|
||||
);
|
||||
assert!(
|
||||
listed_versions
|
||||
.delete_markers()
|
||||
.iter()
|
||||
.any(|marker| marker.key() == Some(key) && marker.version_id() == Some(delete_marker_version_id.as_str())),
|
||||
"ListObjectVersions must expose the delete marker"
|
||||
);
|
||||
|
||||
client
|
||||
.delete_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.version_id(object_version_id)
|
||||
.send()
|
||||
.await
|
||||
.expect("failed to delete the data version");
|
||||
client
|
||||
.delete_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.version_id(delete_marker_version_id)
|
||||
.send()
|
||||
.await
|
||||
.expect("failed to delete the delete marker");
|
||||
|
||||
let remaining_versions = client
|
||||
.list_object_versions()
|
||||
.bucket(bucket)
|
||||
.send()
|
||||
.await
|
||||
.expect("failed to list versions after cleanup");
|
||||
assert!(remaining_versions.versions().is_empty());
|
||||
assert!(remaining_versions.delete_markers().is_empty());
|
||||
|
||||
client
|
||||
.delete_bucket()
|
||||
.bucket(bucket)
|
||||
.send()
|
||||
.await
|
||||
.expect("Deleting every version must remove xl.meta so the bucket can be deleted normally");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_versioning_auto_enabled_with_object_lock() {
|
||||
|
||||
@@ -48,6 +48,12 @@ hotpath-alloc = [
|
||||
"rustfs-filemeta/hotpath-alloc",
|
||||
"rustfs-rio/hotpath-alloc",
|
||||
]
|
||||
hotpath-cpu = [
|
||||
"hotpath",
|
||||
"hotpath/hotpath-cpu",
|
||||
"rustfs-filemeta/hotpath-cpu",
|
||||
"rustfs-rio/hotpath-cpu",
|
||||
]
|
||||
# Exposes shared lifecycle/tier test utilities (MockWarmBackend, fault
|
||||
# injection, xl.meta transition assertions) via `api::tier::test_util`.
|
||||
# Enable only from `[dev-dependencies]` (rustfs/backlog#1148 ilm-6).
|
||||
|
||||
@@ -579,8 +579,26 @@ pub struct BucketMetadataSys {
|
||||
/// Serializes metadata-map commits and their derived cache updates for one
|
||||
/// bucket. Namespace locks, when present, are acquired before this lock.
|
||||
metadata_publish_locks: Arc<MetadataPublishLockRegistry>,
|
||||
/// Deduplicates concurrent lazy loads of one bucket's metadata, so N
|
||||
/// simultaneous cache misses issue a single disk read instead of N.
|
||||
///
|
||||
/// This is the `singleflight` that upstream applies to its own lazy
|
||||
/// `GetConfig`. Without it the namespace *read* lock the load holds is no
|
||||
/// help: read locks are shared, so it excludes concurrent config writers
|
||||
/// but not concurrent readers, and every caller still pays a full
|
||||
/// erasure-set metadata fanout. A separate registry from
|
||||
/// `metadata_publish_locks`, reusing the same per-bucket lock machinery.
|
||||
///
|
||||
/// Lock order: this lock, then the namespace lock, then the publish lock,
|
||||
/// then the metadata map. It is only ever taken as the first of those, so
|
||||
/// it cannot invert against a path that already holds one of the others.
|
||||
lazy_load_locks: Arc<MetadataPublishLockRegistry>,
|
||||
#[cfg(test)]
|
||||
lazy_load_lock_probe: std::sync::atomic::AtomicBool,
|
||||
/// Counts disk loads taken by the lazy `get_config` path, so a test can
|
||||
/// prove concurrent misses collapse into one.
|
||||
#[cfg(test)]
|
||||
lazy_disk_loads: std::sync::atomic::AtomicUsize,
|
||||
/// Buckets recently observed to have no persisted metadata. Serving the
|
||||
/// fabricated default from here (instead of re-reading disk) keeps the
|
||||
/// per-request cost of repeated lookups for such names bounded — without
|
||||
@@ -599,8 +617,13 @@ impl BucketMetadataSys {
|
||||
metadata_publish_locks: Arc::new(MetadataPublishLockRegistry {
|
||||
locks: StdMutex::new(HashMap::new()),
|
||||
}),
|
||||
lazy_load_locks: Arc::new(MetadataPublishLockRegistry {
|
||||
locks: StdMutex::new(HashMap::new()),
|
||||
}),
|
||||
#[cfg(test)]
|
||||
lazy_load_lock_probe: std::sync::atomic::AtomicBool::new(false),
|
||||
#[cfg(test)]
|
||||
lazy_disk_loads: std::sync::atomic::AtomicUsize::new(0),
|
||||
absent_metadata: moka::future::Cache::builder()
|
||||
.max_capacity(ABSENT_BUCKET_METADATA_MAX_ENTRIES)
|
||||
.time_to_live(ABSENT_BUCKET_METADATA_TTL)
|
||||
@@ -615,16 +638,22 @@ impl BucketMetadataSys {
|
||||
}
|
||||
|
||||
fn metadata_publish_lock(&self, bucket: &str) -> Arc<Mutex<MetadataPublishLockState>> {
|
||||
let mut locks = self
|
||||
.metadata_publish_locks
|
||||
.locks
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner());
|
||||
Self::bucket_lock_in(&self.metadata_publish_locks, bucket)
|
||||
}
|
||||
|
||||
/// Per-bucket gate for the lazy `get_config` disk load. See
|
||||
/// [`Self::lazy_load_locks`].
|
||||
fn lazy_load_lock(&self, bucket: &str) -> Arc<Mutex<MetadataPublishLockState>> {
|
||||
Self::bucket_lock_in(&self.lazy_load_locks, bucket)
|
||||
}
|
||||
|
||||
fn bucket_lock_in(registry: &Arc<MetadataPublishLockRegistry>, bucket: &str) -> Arc<Mutex<MetadataPublishLockState>> {
|
||||
let mut locks = registry.locks.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
|
||||
locks.get(bucket).and_then(Weak::upgrade).unwrap_or_else(|| {
|
||||
let lock = Arc::new_cyclic(|lock| {
|
||||
Mutex::new(MetadataPublishLockState {
|
||||
bucket: bucket.to_string(),
|
||||
registry: Arc::downgrade(&self.metadata_publish_locks),
|
||||
registry: Arc::downgrade(registry),
|
||||
lock: lock.clone(),
|
||||
})
|
||||
});
|
||||
@@ -1079,6 +1108,27 @@ impl BucketMetadataSys {
|
||||
return Ok((Arc::new(bm), true));
|
||||
}
|
||||
|
||||
// Collapse concurrent misses for this bucket into one disk load.
|
||||
// Taken before the namespace lock — see `lazy_load_locks` for the
|
||||
// ordering rule.
|
||||
let load_lock = self.lazy_load_lock(bucket);
|
||||
let _load_guard = load_lock.lock_owned().await;
|
||||
|
||||
// Re-check both caches: whoever held the gate before us may have
|
||||
// already answered this exact question, and repeating the fanout
|
||||
// is the whole cost this gate exists to avoid.
|
||||
if let Some(bm) = self.metadata_map.read().await.get(bucket).cloned() {
|
||||
return Ok((bm, true));
|
||||
}
|
||||
if self.absent_metadata.get(bucket).await.is_some() {
|
||||
let mut bm = BucketMetadata::new(bucket);
|
||||
bm.default_timestamps();
|
||||
return Ok((Arc::new(bm), true));
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
self.lazy_disk_loads.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
|
||||
let lock = self.api.new_ns_lock(bucket, bucket).await?;
|
||||
let guard = lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
|
||||
#[cfg(test)]
|
||||
@@ -1431,6 +1481,43 @@ mod tests {
|
||||
use serial_test::serial;
|
||||
use tokio::time::timeout;
|
||||
|
||||
/// Concurrent cache misses for one bucket must collapse into a single disk
|
||||
/// load.
|
||||
///
|
||||
/// The namespace read lock the lazy path already holds does not provide
|
||||
/// this: read locks are shared, so it excludes concurrent config writers
|
||||
/// but not concurrent readers. Without the dedup gate every caller pays its
|
||||
/// own namespace-lock acquisition plus a full erasure-set metadata fanout —
|
||||
/// and the paths that reach `get_config` are per-request, so the multiplier
|
||||
/// is request concurrency.
|
||||
#[tokio::test]
|
||||
async fn concurrent_lazy_loads_of_one_bucket_issue_a_single_disk_read() {
|
||||
use std::sync::atomic::Ordering;
|
||||
|
||||
let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
|
||||
let sys = Arc::new(BucketMetadataSys::new(ecstore));
|
||||
|
||||
// A name with no persisted metadata: every caller misses the map, and
|
||||
// the absent-cache entry does not exist until the first load records it.
|
||||
let bucket = "singleflight-bucket";
|
||||
let waiters = 8;
|
||||
|
||||
let results = futures::future::join_all((0..waiters).map(|_| {
|
||||
let sys = Arc::clone(&sys);
|
||||
async move { sys.get_config(bucket).await.map(|(bm, _)| bm.name.clone()) }
|
||||
}))
|
||||
.await;
|
||||
|
||||
for result in results {
|
||||
assert_eq!(result.expect("every caller must get an answer"), bucket);
|
||||
}
|
||||
assert_eq!(
|
||||
sys.lazy_disk_loads.load(Ordering::Relaxed),
|
||||
1,
|
||||
"concurrent misses for one bucket must share a single disk load"
|
||||
);
|
||||
}
|
||||
|
||||
/// Pins the fail-closed caching contract of the lazy `get_config` path
|
||||
/// and the refresh no-replace rule: fabricated defaults are returned but
|
||||
/// never served by the map-only `get()`, persisted metadata is cached on
|
||||
|
||||
@@ -29,6 +29,7 @@ documentation = "https://docs.rs/rustfs-filemeta/latest/rustfs_filemeta/"
|
||||
default = []
|
||||
hotpath = ["hotpath/hotpath", "hotpath/tokio"]
|
||||
hotpath-alloc = ["hotpath/hotpath-alloc"]
|
||||
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
|
||||
|
||||
[dependencies]
|
||||
hotpath.workspace = true
|
||||
|
||||
@@ -28,6 +28,12 @@ documentation = "https://docs.rs/rustfs-policy/latest/rustfs_policy/"
|
||||
[lints]
|
||||
workspace = true
|
||||
|
||||
[features]
|
||||
default = []
|
||||
hotpath = ["hotpath/hotpath", "hotpath/reqwest-0-13"]
|
||||
hotpath-alloc = ["hotpath/hotpath-alloc"]
|
||||
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
|
||||
|
||||
[dependencies]
|
||||
rustfs-credentials = { workspace = true }
|
||||
rustfs-config = { workspace = true, features = ["opa"] }
|
||||
@@ -49,6 +55,7 @@ moka = { workspace = true, features = ["future"] }
|
||||
async-trait.workspace = true
|
||||
futures.workspace = true
|
||||
pollster.workspace = true
|
||||
hotpath.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
pollster.workspace = true
|
||||
|
||||
@@ -32,10 +32,15 @@ impl Args {
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct AuthZPlugin {
|
||||
client: reqwest::Client,
|
||||
client: OpaHttpClient,
|
||||
args: Args,
|
||||
}
|
||||
|
||||
#[cfg(feature = "hotpath")]
|
||||
type OpaHttpClient = hotpath::wrap::reqwest::Client;
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
type OpaHttpClient = reqwest::Client;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum OpaConfigError {
|
||||
#[error("Missing required env var: {0}")]
|
||||
@@ -141,6 +146,9 @@ impl AuthZPlugin {
|
||||
reqwest::Client::new()
|
||||
});
|
||||
|
||||
#[cfg(feature = "hotpath")]
|
||||
let client = hotpath::http!(client, label = "Policy::OPA");
|
||||
|
||||
Self { client, args: config }
|
||||
}
|
||||
|
||||
|
||||
@@ -32,6 +32,7 @@ workspace = true
|
||||
default = []
|
||||
hotpath = ["hotpath/hotpath", "hotpath/tokio", "hotpath/futures", "hotpath/reqwest-0-13"]
|
||||
hotpath-alloc = ["hotpath/hotpath-alloc"]
|
||||
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
|
||||
|
||||
[dependencies]
|
||||
hotpath.workspace = true
|
||||
|
||||
@@ -75,7 +75,7 @@ const DATA_SCANNER_COMPACT_LEAST_OBJECT: usize = 500;
|
||||
const DATA_SCANNER_COMPACT_AT_CHILDREN: usize = 10000;
|
||||
const DATA_SCANNER_COMPACT_AT_FOLDERS: usize = DATA_SCANNER_COMPACT_AT_CHILDREN / 4;
|
||||
const DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS: usize = 250_000;
|
||||
const SCANNER_LIST_PATH_RAW_TIMEOUT: Duration = Duration::from_secs(60);
|
||||
const SCANNER_LIST_PATH_RAW_STALL_TIMEOUT: Duration = Duration::from_secs(60);
|
||||
const SCANNER_ENTRY_PROGRESS_BATCH: u64 = 32;
|
||||
const SCANNER_ENTRY_PROGRESS_INTERVAL: Duration = Duration::from_secs(30);
|
||||
const DEFAULT_HEAL_OBJECT_SELECT_PROB: u32 = 1024;
|
||||
@@ -100,6 +100,21 @@ static SCANNER_INLINE_HEAL_WARN_ONCE: Once = Once::new();
|
||||
static SCANNER_INLINE_HEAL_METRICS_ONCE: Once = Once::new();
|
||||
static SCANNER_ALERT_METRICS_ONCE: Once = Once::new();
|
||||
|
||||
#[cfg(test)]
|
||||
type ListPathRawTimeoutSnapshot = (bool, Option<Duration>, Option<Duration>);
|
||||
|
||||
fn scanner_abandoned_child_list_options() -> ListPathRawOptions {
|
||||
// A complete heal walk scales with bucket size and may legitimately take
|
||||
// longer than a fixed wall-clock budget. Keep the total duration unbounded;
|
||||
// Retain the scanner's per-read stall budget and keep cancellation controlled
|
||||
// by the scanner cycle token.
|
||||
ListPathRawOptions {
|
||||
skip_walkdir_total_timeout: true,
|
||||
walkdir_stall_timeout: Some(SCANNER_LIST_PATH_RAW_STALL_TIMEOUT),
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
pub fn data_usage_update_dir_cycles() -> u32 {
|
||||
rustfs_utils::get_env_u32(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, DATA_USAGE_UPDATE_DIR_CYCLES)
|
||||
}
|
||||
@@ -1281,6 +1296,8 @@ pub struct FolderScanner {
|
||||
skip_heal: Arc<std::sync::atomic::AtomicBool>,
|
||||
local_disk: Arc<Disk>,
|
||||
pending_heals_changed: bool,
|
||||
#[cfg(test)]
|
||||
list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
|
||||
}
|
||||
|
||||
impl FolderScanner {
|
||||
@@ -2410,75 +2427,79 @@ impl FolderScanner {
|
||||
let bucket_clone = bucket.clone();
|
||||
let prefix_clone = prefix.clone();
|
||||
let child_ctx_clone = child_ctx.clone();
|
||||
#[cfg(test)]
|
||||
let list_path_raw_options_observer = self.list_path_raw_options_observer.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = list_path_raw(
|
||||
child_ctx_clone.clone(),
|
||||
ListPathRawOptions {
|
||||
disks,
|
||||
bucket: bucket_clone.clone(),
|
||||
path: prefix_clone.clone(),
|
||||
recursive: true,
|
||||
report_not_found: true,
|
||||
min_disks: disks_quorum,
|
||||
walkdir_timeout: Some(SCANNER_LIST_PATH_RAW_TIMEOUT),
|
||||
walkdir_stall_timeout: Some(SCANNER_LIST_PATH_RAW_TIMEOUT),
|
||||
agreed: Some(Box::new(move |entry: MetaCacheEntry| {
|
||||
let entry_name = entry.name.clone();
|
||||
let agreed_tx = agreed_tx.clone();
|
||||
Box::pin(async move {
|
||||
if let Err(e) = agreed_tx.send(entry_name).await {
|
||||
error!(
|
||||
target: "rustfs::scanner::folder",
|
||||
event = EVENT_SCANNER_FOLDER_STATE,
|
||||
component = LOG_COMPONENT_SCANNER,
|
||||
subsystem = LOG_SUBSYSTEM_FOLDER,
|
||||
entry = %entry.name,
|
||||
state = "list_path_agreed_send_failed",
|
||||
error = %e,
|
||||
"Scanner list_path_raw agreed callback failed"
|
||||
);
|
||||
}
|
||||
})
|
||||
})),
|
||||
partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
|
||||
let partial_tx = partial_tx.clone();
|
||||
Box::pin(async move {
|
||||
if let Err(e) = partial_tx.send(entries).await {
|
||||
error!(
|
||||
target: "rustfs::scanner::folder",
|
||||
event = EVENT_SCANNER_FOLDER_STATE,
|
||||
component = LOG_COMPONENT_SCANNER,
|
||||
subsystem = LOG_SUBSYSTEM_FOLDER,
|
||||
state = "list_path_partial_send_failed",
|
||||
error = %e,
|
||||
"Scanner list_path_raw partial callback failed"
|
||||
);
|
||||
}
|
||||
})
|
||||
})),
|
||||
finished: Some(Box::new(move |errs: &[Option<DiskError>]| {
|
||||
let finished_tx = finished_tx.clone();
|
||||
let errs_clone = errs.to_vec();
|
||||
Box::pin(async move {
|
||||
if let Err(e) = finished_tx.send(errs_clone).await {
|
||||
error!(
|
||||
target: "rustfs::scanner::folder",
|
||||
event = EVENT_SCANNER_FOLDER_STATE,
|
||||
component = LOG_COMPONENT_SCANNER,
|
||||
subsystem = LOG_SUBSYSTEM_FOLDER,
|
||||
state = "list_path_finished_send_failed",
|
||||
error = %e,
|
||||
"Scanner list_path_raw finished callback failed"
|
||||
);
|
||||
}
|
||||
})
|
||||
})),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
{
|
||||
let options = ListPathRawOptions {
|
||||
disks,
|
||||
bucket: bucket_clone.clone(),
|
||||
path: prefix_clone.clone(),
|
||||
recursive: true,
|
||||
report_not_found: true,
|
||||
min_disks: disks_quorum,
|
||||
agreed: Some(Box::new(move |entry: MetaCacheEntry| {
|
||||
let entry_name = entry.name.clone();
|
||||
let agreed_tx = agreed_tx.clone();
|
||||
Box::pin(async move {
|
||||
if let Err(e) = agreed_tx.send(entry_name).await {
|
||||
error!(
|
||||
target: "rustfs::scanner::folder",
|
||||
event = EVENT_SCANNER_FOLDER_STATE,
|
||||
component = LOG_COMPONENT_SCANNER,
|
||||
subsystem = LOG_SUBSYSTEM_FOLDER,
|
||||
entry = %entry.name,
|
||||
state = "list_path_agreed_send_failed",
|
||||
error = %e,
|
||||
"Scanner list_path_raw agreed callback failed"
|
||||
);
|
||||
}
|
||||
})
|
||||
})),
|
||||
partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
|
||||
let partial_tx = partial_tx.clone();
|
||||
Box::pin(async move {
|
||||
if let Err(e) = partial_tx.send(entries).await {
|
||||
error!(
|
||||
target: "rustfs::scanner::folder",
|
||||
event = EVENT_SCANNER_FOLDER_STATE,
|
||||
component = LOG_COMPONENT_SCANNER,
|
||||
subsystem = LOG_SUBSYSTEM_FOLDER,
|
||||
state = "list_path_partial_send_failed",
|
||||
error = %e,
|
||||
"Scanner list_path_raw partial callback failed"
|
||||
);
|
||||
}
|
||||
})
|
||||
})),
|
||||
finished: Some(Box::new(move |errs: &[Option<DiskError>]| {
|
||||
let finished_tx = finished_tx.clone();
|
||||
let errs_clone = errs.to_vec();
|
||||
Box::pin(async move {
|
||||
if let Err(e) = finished_tx.send(errs_clone).await {
|
||||
error!(
|
||||
target: "rustfs::scanner::folder",
|
||||
event = EVENT_SCANNER_FOLDER_STATE,
|
||||
component = LOG_COMPONENT_SCANNER,
|
||||
subsystem = LOG_SUBSYSTEM_FOLDER,
|
||||
state = "list_path_finished_send_failed",
|
||||
error = %e,
|
||||
"Scanner list_path_raw finished callback failed"
|
||||
);
|
||||
}
|
||||
})
|
||||
})),
|
||||
..scanner_abandoned_child_list_options()
|
||||
};
|
||||
#[cfg(test)]
|
||||
if let Some(observer) = list_path_raw_options_observer {
|
||||
let _ = observer.send((
|
||||
options.skip_walkdir_total_timeout,
|
||||
options.walkdir_timeout,
|
||||
options.walkdir_stall_timeout,
|
||||
));
|
||||
}
|
||||
if let Err(e) = list_path_raw(child_ctx_clone.clone(), options).await {
|
||||
if is_missing_path_disk_error(&e) {
|
||||
debug!(
|
||||
target: "rustfs::scanner::folder",
|
||||
@@ -2820,6 +2841,8 @@ pub async fn scan_data_folder(
|
||||
skip_heal,
|
||||
local_disk,
|
||||
pending_heals_changed: false,
|
||||
#[cfg(test)]
|
||||
list_path_raw_options_observer: None,
|
||||
};
|
||||
|
||||
// Check if context is cancelled
|
||||
@@ -3026,6 +3049,7 @@ mod tests {
|
||||
skip_heal: Arc::new(AtomicBool::new(false)),
|
||||
local_disk: disk,
|
||||
pending_heals_changed: false,
|
||||
list_path_raw_options_observer: None,
|
||||
};
|
||||
|
||||
(scanner, temp_dir)
|
||||
@@ -4210,6 +4234,8 @@ mod tests {
|
||||
scanner.heal_object_select = 1;
|
||||
scanner.disks = disks;
|
||||
scanner.disks_quorum = 2;
|
||||
let (options_tx, mut options_rx) = mpsc::unbounded_channel();
|
||||
scanner.list_path_raw_options_observer = Some(options_tx);
|
||||
scanner.old_cache.replace(
|
||||
&format!("{bucket}/{object}"),
|
||||
bucket,
|
||||
@@ -4231,6 +4257,11 @@ mod tests {
|
||||
.expect("scan_folder should not hang after list_path_raw finishes")
|
||||
.expect("scan_folder should finish successfully");
|
||||
|
||||
let observed_options = tokio::time::timeout(Duration::from_secs(1), options_rx.recv())
|
||||
.await
|
||||
.expect("abandoned-child listing options should be observed promptly")
|
||||
.expect("abandoned-child listing options channel should remain open");
|
||||
assert_eq!(observed_options, (true, None, Some(SCANNER_LIST_PATH_RAW_STALL_TIMEOUT)));
|
||||
let root = scanner
|
||||
.new_cache
|
||||
.checked_flatten(bucket)
|
||||
|
||||
@@ -62,14 +62,24 @@ hotpath = [
|
||||
"hotpath/reqwest-0-13",
|
||||
"rustfs-ecstore/hotpath",
|
||||
"rustfs-filemeta/hotpath",
|
||||
"rustfs-policy/hotpath",
|
||||
"rustfs-rio/hotpath",
|
||||
]
|
||||
hotpath-alloc = [
|
||||
"hotpath/hotpath-alloc",
|
||||
"rustfs-ecstore/hotpath-alloc",
|
||||
"rustfs-filemeta/hotpath-alloc",
|
||||
"rustfs-policy/hotpath-alloc",
|
||||
"rustfs-rio/hotpath-alloc",
|
||||
]
|
||||
hotpath-cpu = [
|
||||
"hotpath",
|
||||
"hotpath/hotpath-cpu",
|
||||
"rustfs-ecstore/hotpath-cpu",
|
||||
"rustfs-filemeta/hotpath-cpu",
|
||||
"rustfs-policy/hotpath-cpu",
|
||||
"rustfs-rio/hotpath-cpu",
|
||||
]
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
|
||||
Reference in New Issue
Block a user