mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-29 17:48:58 +00:00
Merge remote-tracking branch 'origin/main' into cxymds/fix-1358-opaque-transition-metadata
This commit is contained in:
马登山
@@ -141,7 +141,7 @@ jobs:
|
||||
name: Test and Lint
|
||||
if: github.event_name != 'pull_request' || github.event.action != 'closed'
|
||||
runs-on: sm-standard-4
|
||||
timeout-minutes: 60
|
||||
timeout-minutes: 90
|
||||
env:
|
||||
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
|
||||
steps:
|
||||
|
||||
Generated
+29
-30
@@ -1703,9 +1703,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "camino"
|
||||
version = "1.2.4"
|
||||
version = "1.2.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5f2d30e4173c4026932d51d31d6b0613b1fd3014bf3f9f8943d4ba139c437ba0"
|
||||
checksum = "bb1307f12aa967b5a58416e87b3653360e0fd614a016b6e970db08fecbb1b80d"
|
||||
dependencies = [
|
||||
"serde_core",
|
||||
]
|
||||
@@ -3625,13 +3625,13 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "displaydoc"
|
||||
version = "0.2.6"
|
||||
version = "0.2.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f"
|
||||
checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.119",
|
||||
"syn 3.0.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -3941,16 +3941,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"windows-sys 0.52.0",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "event-listener"
|
||||
version = "5.4.1"
|
||||
version = "5.4.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e13b66accf52311f30a0db42147dadea9850cb48cd070028831ae5f5d4b856ab"
|
||||
checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2"
|
||||
dependencies = [
|
||||
"concurrent-queue",
|
||||
"parking",
|
||||
"pin-project-lite",
|
||||
]
|
||||
@@ -5361,7 +5360,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
|
||||
dependencies = [
|
||||
"hermit-abi",
|
||||
"libc",
|
||||
"windows-sys 0.52.0",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -6110,9 +6109,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "metrique"
|
||||
version = "0.1.28"
|
||||
version = "0.1.29"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "aa466af30a9fe0b1db1dae097e0ce7ddac4b666b1d3a9f5c682e889272511dd8"
|
||||
checksum = "d2e394c63e2d1a30aeb3b9392ecf3439d8475d2df810a8f4f6e66d6866754017"
|
||||
dependencies = [
|
||||
"itoa",
|
||||
"jiff",
|
||||
@@ -6140,9 +6139,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "metrique-macro"
|
||||
version = "0.1.19"
|
||||
version = "0.1.20"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1f5febfaf14fea234b60e0ad43c728db274033893a38d0fa1f87d9b7056d3d61"
|
||||
checksum = "786df1fd0abebd0db685f7e9a353c78756d4b370fb98a52376c2015fa55f141f"
|
||||
dependencies = [
|
||||
"Inflector",
|
||||
"darling 0.23.0",
|
||||
@@ -6169,9 +6168,9 @@ checksum = "2faca4e4480069ff02b1763b3b79f5cec7e8628e24d9dc5b6073f53d2577a4d9"
|
||||
|
||||
[[package]]
|
||||
name = "metrique-writer"
|
||||
version = "0.1.24"
|
||||
version = "0.1.25"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "124326a2ac4c4f61562fa4d071735a1c463f9ba0317d1564f75dc01313dc12d7"
|
||||
checksum = "82cdde44d241dab7fc8b7a32e0eb5dae6cd28f8de80b59f9a1e9f2f0b05e485e"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"crossbeam-queue",
|
||||
@@ -6190,9 +6189,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "metrique-writer-core"
|
||||
version = "0.1.18"
|
||||
version = "0.1.19"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "55b5bbb6d88bde29f6ed74a574cbe49e51ea0c36cccc7e63eee2040db5675b15"
|
||||
checksum = "e57379b7ee2272efaeaaa6de062503563e57333b24aadc7f2255b3d602899e8b"
|
||||
dependencies = [
|
||||
"derive-where",
|
||||
"itertools 0.14.0",
|
||||
@@ -8097,7 +8096,7 @@ dependencies = [
|
||||
"once_cell",
|
||||
"socket2",
|
||||
"tracing",
|
||||
"windows-sys 0.52.0",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -10225,7 +10224,7 @@ dependencies = [
|
||||
"errno",
|
||||
"libc",
|
||||
"linux-raw-sys",
|
||||
"windows-sys 0.52.0",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -10298,7 +10297,7 @@ dependencies = [
|
||||
"security-framework",
|
||||
"security-framework-sys",
|
||||
"webpki-root-certs",
|
||||
"windows-sys 0.52.0",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -10355,7 +10354,7 @@ checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
|
||||
[[package]]
|
||||
name = "s3s"
|
||||
version = "0.14.1"
|
||||
source = "git+https://github.com/cxymds/s3s.git?rev=afa1796ec64cd1e1c78dfee46c4b4b72bbc7b39d#afa1796ec64cd1e1c78dfee46c4b4b72bbc7b39d"
|
||||
source = "git+https://github.com/cxymds/s3s.git?rev=fe3941d91fa1c69956f209a9145995c9f0235bff#fe3941d91fa1c69956f209a9145995c9f0235bff"
|
||||
dependencies = [
|
||||
"arc-swap",
|
||||
"arrayvec",
|
||||
@@ -10461,9 +10460,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "schemars"
|
||||
version = "1.2.1"
|
||||
version = "1.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a2b42f36aa1cd011945615b92222f6bf73c599a102a300334cd7f8dbeec726cc"
|
||||
checksum = "687274d293b6cdc6e73e0fee520bf2049650090d7164f87672d212a3c530cf4a"
|
||||
dependencies = [
|
||||
"dyn-clone",
|
||||
"ref-cast",
|
||||
@@ -10696,7 +10695,7 @@ dependencies = [
|
||||
"indexmap 1.9.3",
|
||||
"indexmap 2.14.0",
|
||||
"schemars 0.9.0",
|
||||
"schemars 1.2.1",
|
||||
"schemars 1.2.2",
|
||||
"serde_core",
|
||||
"serde_json",
|
||||
"serde_with_macros",
|
||||
@@ -11498,10 +11497,10 @@ 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.52.0",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -11876,9 +11875,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "toml_parser"
|
||||
version = "1.1.2+spec-1.1.0"
|
||||
version = "1.1.3+spec-1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a2abe9b86193656635d2411dc43050282ca48aa31c2451210f4202550afb7526"
|
||||
checksum = "1d38ac1cf9b95face32296c0a3ede1fdc270627c9d9c02a7274dd6d960dc4d56"
|
||||
dependencies = [
|
||||
"winnow",
|
||||
]
|
||||
@@ -12602,7 +12601,7 @@ version = "0.1.11"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
|
||||
dependencies = [
|
||||
"windows-sys 0.52.0",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
+1
-1
@@ -288,7 +288,7 @@ redis = { version = "1.4.1" }
|
||||
rustix = { version = "1.1.4" }
|
||||
rust-embed = { version = "8.12.0" }
|
||||
rustc-hash = { version = "2.1.3" }
|
||||
s3s = { git = "https://github.com/cxymds/s3s.git", rev = "afa1796ec64cd1e1c78dfee46c4b4b72bbc7b39d" }
|
||||
s3s = { git = "https://github.com/cxymds/s3s.git", rev = "fe3941d91fa1c69956f209a9145995c9f0235bff" }
|
||||
serial_test = "4.0.1"
|
||||
shadow-rs = { default-features = false, version = "2.0.0" }
|
||||
siphasher = "1.0.3"
|
||||
|
||||
@@ -45,10 +45,10 @@
|
||||
//! * Parity reconstruction: one data disk is taken offline
|
||||
//! (`take_disk_offline`) and the SAME object matrix is GET both ways while
|
||||
//! the EC 2+2 set rebuilds each large object from the surviving shards. The
|
||||
//! codec-streaming reader gate never inspects drive health, so the codec
|
||||
//! fast path is exercised end-to-end through reconstruction; the test
|
||||
//! asserts byte- and header-equality vs the legacy path AND that the codec
|
||||
//! phase never fell back to a duplex pipe while reconstructing.
|
||||
//! eager first/single-part setup may keep its conservative whole-request
|
||||
//! fallback when shard placement makes codec streaming unsafe, so this phase
|
||||
//! asserts byte- and header-equality vs the legacy path rather than requiring
|
||||
//! zero duplex fallbacks under degraded drive health.
|
||||
//! * Missing object: a GET for an absent key is compared across both phases
|
||||
//! to prove the error semantics (HTTP status + S3 error code) are identical
|
||||
//! — the codec env must not perturb the NoSuchKey negative path.
|
||||
@@ -475,15 +475,14 @@ mod tests {
|
||||
"ranged GET length diverged with codec streaming enabled"
|
||||
);
|
||||
|
||||
// ---- Phase B degraded: the same reconstruction, now on the codec path ----
|
||||
// Re-run the reconstruction A/B with the codec-streaming gates still
|
||||
// open. The reader gate decision is independent of drive health (it
|
||||
// never inspects disk state), so the codec fast path is exercised
|
||||
// end-to-end while the EC set rebuilds each large object from the
|
||||
// surviving shards — this is a real codec-vs-legacy reconstruction test,
|
||||
// not legacy-vs-legacy. Snapshot the duplex count first (the range GET
|
||||
// above already used the duplex path) so we can measure only the markers
|
||||
// these degraded codec GETs add.
|
||||
// ---- Phase B degraded: the same reconstruction, with codec gates open ----
|
||||
// Re-run the reconstruction A/B with codec-streaming enabled. If eager
|
||||
// first/single-part setup cannot prove the codec path is safe for the
|
||||
// surviving shards, the implementation intentionally preserves the
|
||||
// whole-request legacy fallback; later multipart parts can degrade in
|
||||
// place. This phase verifies parity-reconstructed bytes and headers,
|
||||
// while the healthy phase above remains the strict zero-duplex path
|
||||
// confirmation.
|
||||
let dup_codec_before_degraded = count_marker(&codec_log, DUPLEX_MARKER);
|
||||
harness.take_disk_offline(0)?;
|
||||
let mut codec_degraded: BTreeMap<String, GetView> = BTreeMap::new();
|
||||
@@ -511,16 +510,11 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
// Path confirmation under reconstruction: the codec fast path must have
|
||||
// served the reconstructed large objects without ever falling back to
|
||||
// the legacy duplex pipe. Without this, the equivalence above could be
|
||||
// legacy-vs-legacy and prove nothing about codec reconstruction.
|
||||
// Keep degraded duplex markers as diagnostic evidence only: eager setup
|
||||
// may fall back before streaming when shard safety cannot be proven.
|
||||
sleep(Duration::from_millis(300)).await;
|
||||
let dup_codec_degraded = count_marker(&codec_log, DUPLEX_MARKER).saturating_sub(dup_codec_before_degraded);
|
||||
assert_eq!(
|
||||
dup_codec_degraded, 0,
|
||||
"codec phase created {dup_codec_degraded} duplex pipe(s) while reconstructing large objects with disk0 offline; the codec fast path was not exercised under degraded reads (see {codec_log})"
|
||||
);
|
||||
info!(dup_codec_degraded, "codec phase degraded-read legacy duplex marker count");
|
||||
|
||||
info!(
|
||||
objects = baseline.len(),
|
||||
|
||||
@@ -7240,6 +7240,9 @@ mod transition_source_identity_matrix_tests {
|
||||
changed.metadata.insert("etag".to_string(), format!("changed-{index}"));
|
||||
}
|
||||
}
|
||||
// Replace the object version list so VersionId drift removes the
|
||||
// accepted source version instead of appending a second version.
|
||||
changed.fresh = true;
|
||||
for disk in &disk_stores {
|
||||
disk.write_metadata("", bucket, &object, changed.clone())
|
||||
.await
|
||||
|
||||
@@ -2016,11 +2016,15 @@ mod metadata_cache_tests {
|
||||
fi.size = 1;
|
||||
fi.erasure.index = 1;
|
||||
fi.metadata.insert("etag".to_string(), "etag-1".to_string());
|
||||
fi.add_object_part(1, "part-etag".to_string(), 1, fi.mod_time, 1, None, None);
|
||||
fi
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn get_object_with_fileinfo_rejects_positive_size_without_parts() {
|
||||
let mut fi = valid_test_fileinfo("object");
|
||||
fi.parts.clear();
|
||||
|
||||
let mut output = Vec::new();
|
||||
let err = SetDisks::get_object_with_fileinfo(
|
||||
"bucket",
|
||||
@@ -2028,7 +2032,7 @@ mod metadata_cache_tests {
|
||||
0,
|
||||
1,
|
||||
&mut output,
|
||||
valid_test_fileinfo("object"),
|
||||
fi,
|
||||
Vec::new(),
|
||||
&[],
|
||||
0,
|
||||
@@ -2119,12 +2123,6 @@ mod metadata_cache_tests {
|
||||
|
||||
let mut invalid_erasure = valid_test_fileinfo(object);
|
||||
invalid_erasure.erasure.block_size = 0;
|
||||
invalid_erasure.parts.push(ObjectPartInfo {
|
||||
number: 1,
|
||||
size: 1,
|
||||
actual_size: 1,
|
||||
..Default::default()
|
||||
});
|
||||
let err = SetDisks::get_object_with_fileinfo(
|
||||
bucket,
|
||||
object,
|
||||
@@ -2159,6 +2157,7 @@ mod metadata_cache_tests {
|
||||
let object = "empty";
|
||||
let mut fi = valid_test_fileinfo(object);
|
||||
fi.size = 0;
|
||||
fi.parts.clear();
|
||||
|
||||
let mut output = Vec::new();
|
||||
SetDisks::get_object_with_fileinfo(
|
||||
@@ -2191,12 +2190,6 @@ mod metadata_cache_tests {
|
||||
let mut fi = valid_test_fileinfo(object);
|
||||
fi.erasure.block_size = 1;
|
||||
fi.erasure.distribution = vec![1, 2, 3, 4];
|
||||
fi.parts.push(ObjectPartInfo {
|
||||
number: 1,
|
||||
size: 1,
|
||||
actual_size: 1,
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let mut output = Vec::new();
|
||||
let err = SetDisks::get_object_with_fileinfo(
|
||||
|
||||
@@ -25,6 +25,9 @@ keywords = ["file-system", "notification", "real-time", "rustfs", "Minio"]
|
||||
categories = ["web-programming", "development-tools", "filesystem"]
|
||||
documentation = "https://docs.rs/rustfs-notify/latest/rustfs_notify/"
|
||||
|
||||
[features]
|
||||
demo-examples = []
|
||||
|
||||
[dependencies]
|
||||
rustfs-config = { workspace = true, features = ["notify", "server-config-model"] }
|
||||
rustfs-ecstore = { workspace = true }
|
||||
@@ -71,6 +74,18 @@ workspace = true
|
||||
[lib]
|
||||
doctest = false
|
||||
|
||||
[[example]]
|
||||
name = "full_demo"
|
||||
required-features = ["demo-examples"]
|
||||
|
||||
[[example]]
|
||||
name = "full_demo_one"
|
||||
required-features = ["demo-examples"]
|
||||
|
||||
[[example]]
|
||||
name = "webhook"
|
||||
required-features = ["demo-examples"]
|
||||
|
||||
[[bench]]
|
||||
name = "snapshot_mode_scan"
|
||||
harness = false
|
||||
|
||||
@@ -40,7 +40,7 @@ allow-git = [
|
||||
# SigV4 payload-checksum fix until it is available in a crates.io release.
|
||||
# owner: marshawcoco review: 2026-10
|
||||
"https://github.com/s3s-project/s3s.git",
|
||||
# Presigned URL expiry validation pending upstream submission.
|
||||
# Presigned expiry and constant-time authentication fixes pending upstream merge.
|
||||
# owner: cxymds review: 2026-10
|
||||
"https://github.com/cxymds/s3s.git",
|
||||
"https://github.com/apache/datafusion.git",
|
||||
|
||||
@@ -33,7 +33,7 @@ use crate::admin::runtime_sources::{
|
||||
};
|
||||
use crate::admin::storage_api::access::{ReqInfo, authorize_request, spawn_traced};
|
||||
use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions};
|
||||
use crate::auth::{check_key_valid, get_session_token};
|
||||
use crate::auth::{check_key_valid, constant_time_eq, get_session_token};
|
||||
use crate::error::ApiError;
|
||||
use crate::license::license_check;
|
||||
use crate::server::{
|
||||
@@ -905,7 +905,9 @@ fn validate_object_lambda_response_auth_headers(headers: &HeaderMap, output_rout
|
||||
.and_then(|value| value.to_str().ok())
|
||||
.map(str::trim);
|
||||
|
||||
if route == Some(output_route) && token == Some(output_token) {
|
||||
if route.is_some_and(|route| constant_time_eq(route, output_route))
|
||||
&& token.is_some_and(|token| constant_time_eq(token, output_token))
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
@@ -4351,6 +4353,39 @@ mod tests {
|
||||
let err = validate_object_lambda_response_auth_headers(&mismatched, "route-123", "token-456")
|
||||
.expect_err("mismatched auth headers should fail");
|
||||
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
|
||||
|
||||
for (route, token) in [
|
||||
("Route-123", "token-456"),
|
||||
("route-124", "token-456"),
|
||||
("route-1234", "token-456"),
|
||||
("route-123", "Token-456"),
|
||||
("route-123", "token-457"),
|
||||
("route-123", "token-4567"),
|
||||
] {
|
||||
let mut headers = HeaderMap::new();
|
||||
headers.insert(
|
||||
"x-amz-request-route",
|
||||
HeaderValue::try_from(route).expect("test route must be a valid header"),
|
||||
);
|
||||
headers.insert(
|
||||
"x-amz-request-token",
|
||||
HeaderValue::try_from(token).expect("test token must be a valid header"),
|
||||
);
|
||||
assert!(
|
||||
validate_object_lambda_response_auth_headers(&headers, "route-123", "token-456").is_err(),
|
||||
"first-byte, last-byte, and length mismatches must fail: {route}/{token}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn object_lambda_auth_headers_use_constant_time_helper() {
|
||||
let source = include_str!("router.rs");
|
||||
let production = source.split_once("#[cfg(test)]").map_or(source, |(production, _)| production);
|
||||
assert!(!production.contains("route == Some(output_route)"));
|
||||
assert!(!production.contains("token == Some(output_token)"));
|
||||
assert!(production.contains("constant_time_eq(route, output_route)"));
|
||||
assert!(production.contains("constant_time_eq(token, output_token)"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -198,13 +198,9 @@ async fn data_usage_endpoint_serves_snapshot_without_live_listing() {
|
||||
first_info.total_free_capacity
|
||||
);
|
||||
assert_eq!(
|
||||
(first_info.total_capacity, first_info.total_free_capacity, first_info.total_used_capacity,),
|
||||
(
|
||||
second_info.total_capacity,
|
||||
second_info.total_free_capacity,
|
||||
second_info.total_used_capacity,
|
||||
),
|
||||
"repeated data usage requests must report stable capacity values"
|
||||
second_info.total_used_capacity,
|
||||
second_info.total_capacity.saturating_sub(second_info.total_free_capacity),
|
||||
"server used capacity must stay internally consistent on repeated requests"
|
||||
);
|
||||
|
||||
// The endpoint must serve the seeded snapshot numbers, not recomputed ones.
|
||||
|
||||
@@ -60,6 +60,7 @@ use uuid::Uuid;
|
||||
static GLOBAL_ENV: OnceLock<(Vec<PathBuf>, Arc<ECStore>)> = OnceLock::new();
|
||||
static INIT: Once = Once::new();
|
||||
const TRANSITION_WAIT_TIMEOUT: Duration = Duration::from_secs(15);
|
||||
const RESTORE_COPY_BACK_WAIT_TIMEOUT: Duration = Duration::from_secs(60);
|
||||
const ENV_GET_CODEC_STREAMING_ENABLE: &str = "RUSTFS_GET_CODEC_STREAMING_ENABLE";
|
||||
const ENV_GET_CODEC_STREAMING_ROLLOUT: &str = "RUSTFS_GET_CODEC_STREAMING_ROLLOUT";
|
||||
const ENV_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED";
|
||||
@@ -169,6 +170,34 @@ async fn upload_test_object(ecstore: &Arc<ECStore>, bucket: &str, object: &str,
|
||||
.expect("Failed to upload test object")
|
||||
}
|
||||
|
||||
async fn transition_uploaded_object_directly(
|
||||
ecstore: &Arc<ECStore>,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
tier_name: &str,
|
||||
uploaded: &ObjectInfo,
|
||||
) -> ObjectInfo {
|
||||
let transition_opts = ObjectOptions {
|
||||
transition: lifecycle::lifecycle_contract::TransitionOptions {
|
||||
status: lifecycle::lifecycle_contract::TRANSITION_PENDING.to_string(),
|
||||
tier: tier_name.to_string(),
|
||||
etag: uploaded.etag.clone().unwrap_or_default(),
|
||||
..Default::default()
|
||||
},
|
||||
version_id: uploaded.version_id.map(|version| version.to_string()),
|
||||
versioned: uploaded.version_id.is_some(),
|
||||
mod_time: uploaded.mod_time,
|
||||
..Default::default()
|
||||
};
|
||||
ecstore
|
||||
.transition_object(bucket, object, &transition_opts)
|
||||
.await
|
||||
.expect("Failed to transition object directly");
|
||||
wait_for_transition(ecstore, bucket, object, TRANSITION_WAIT_TIMEOUT)
|
||||
.await
|
||||
.expect("object should transition before restore assertions")
|
||||
}
|
||||
|
||||
async fn set_bucket_lifecycle_transition_with_tier(
|
||||
bucket_name: &str,
|
||||
storage_class: &str,
|
||||
@@ -310,6 +339,45 @@ async fn wait_for_transition(ecstore: &Arc<ECStore>, bucket: &str, object: &str,
|
||||
}
|
||||
}
|
||||
|
||||
async fn wait_for_restore_completion(
|
||||
ecstore: &Arc<ECStore>,
|
||||
backend: &MockWarmBackend,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
timeout: Duration,
|
||||
) -> Result<ObjectInfo, String> {
|
||||
let deadline = tokio::time::Instant::now() + timeout;
|
||||
let mut last_state = None;
|
||||
|
||||
loop {
|
||||
if tokio::time::Instant::now() >= deadline {
|
||||
let tier_gets = backend.get_count().await;
|
||||
let op_log = backend.op_log().await;
|
||||
return Err(format!(
|
||||
"restore copy-back should complete within {timeout:?}; tier_gets={tier_gets}, op_log={op_log:?}; last observed state: {}",
|
||||
last_state.unwrap_or_else(|| "no object info observed".to_string())
|
||||
));
|
||||
}
|
||||
|
||||
match (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await {
|
||||
Ok(info) => {
|
||||
if !info.restore_ongoing && info.restore_expires.is_some() {
|
||||
return Ok(info);
|
||||
}
|
||||
last_state = Some(format!(
|
||||
"restore_ongoing={}, restore_expires={:?}, transitioned_status={}",
|
||||
info.restore_ongoing, info.restore_expires, info.transitioned_object.status
|
||||
));
|
||||
}
|
||||
Err(err) => {
|
||||
last_state = Some(format!("get_object_info failed: {err}"));
|
||||
}
|
||||
}
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
}
|
||||
|
||||
// SAFETY: this helper is used only by `#[serial]` tests and runs under the single-threaded Tokio
|
||||
// runtime (`worker_threads = 1`), so no concurrent test can mutate process environment during the
|
||||
// `env::set_var` / `env::remove_var` window.
|
||||
@@ -2026,10 +2094,10 @@ async fn put_bucket_lifecycle_configuration_rejects_zero_day_expiration() {
|
||||
|
||||
/// backlog#1148 ilm-8: the RestoreObject API surface on a transitioned object.
|
||||
///
|
||||
/// POST restore(days=1) is accepted and immediately flips the object to
|
||||
/// `x-amz-restore: ongoing-request="true"` (the mock tier's injected GET
|
||||
/// latency keeps the background copy-back in flight); a second POST during
|
||||
/// that window is rejected with 409 `RestoreAlreadyInProgress`; once the
|
||||
/// POST restore(days=1) is accepted and flips the object to
|
||||
/// `x-amz-restore: ongoing-request="true"` while the mock tier GET barrier
|
||||
/// proves the background copy-back has reached the remote read; a second POST
|
||||
/// during that window is rejected with 409 `RestoreAlreadyInProgress`; once the
|
||||
/// copy-back completes the object reports `ongoing-request="false"` with a
|
||||
/// future expiry-date; and a full GET is then served from the local restored
|
||||
/// copy (the mock tier records no further `get` calls).
|
||||
@@ -2050,29 +2118,17 @@ async fn restore_object_usecase_reports_ongoing_conflict_and_completion() {
|
||||
let backend = register_mock_tier(&tier_name).await;
|
||||
|
||||
let bucket = format!("test-api-restore-{}", &Uuid::new_v4().simple().to_string()[..8]);
|
||||
// Must live under the `test/` prefix: `set_bucket_lifecycle_transition_with_tier`
|
||||
// scopes the transition rule to `<Filter><Prefix>test/</Prefix>`, so an object
|
||||
// outside it never matches, is never enqueued, and never transitions — the
|
||||
// setup `wait_for_transition` would then time out before the restore assertions.
|
||||
// Keep the object under the shared ILM test prefix even though this setup
|
||||
// transitions it directly; it keeps diagnostics aligned with sibling tests.
|
||||
let object = "test/restore/api-object.bin";
|
||||
let payload: Vec<u8> = (0..128 * 1024).map(|i| (i % 251) as u8).collect();
|
||||
|
||||
create_test_bucket(&ecstore, bucket.as_str()).await;
|
||||
set_bucket_lifecycle_transition_with_tier(bucket.as_str(), &tier_name)
|
||||
.await
|
||||
.expect("Failed to set lifecycle configuration");
|
||||
let _ = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
|
||||
let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
|
||||
let _ = transition_uploaded_object_directly(&ecstore, bucket.as_str(), object, &tier_name, &uploaded).await;
|
||||
backend.clear_op_log().await;
|
||||
|
||||
lifecycle::bucket_lifecycle_ops::enqueue_transition_for_existing_objects(ecstore.clone(), bucket.as_str())
|
||||
.await
|
||||
.expect("Failed to enqueue transitioned object");
|
||||
let _ = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
|
||||
.await
|
||||
.expect("object should transition before the restore API runs");
|
||||
|
||||
// Slow the tier GET so the background copy-back stays in flight long
|
||||
// enough to observe the ongoing state and the conflict rejection.
|
||||
backend.set_latency(Some(Duration::from_millis(1500))).await;
|
||||
let get_barrier = backend.arm_get_barrier().await;
|
||||
|
||||
let restore_request = || RestoreRequest {
|
||||
days: Some(1),
|
||||
@@ -2096,15 +2152,18 @@ async fn restore_object_usecase_reports_ongoing_conflict_and_completion() {
|
||||
.await
|
||||
.expect("restore request should be accepted");
|
||||
|
||||
// The accepted restore is immediately visible as ongoing (the metadata is
|
||||
// written synchronously before the copy-back is spawned).
|
||||
get_barrier.wait_until_paused().await;
|
||||
|
||||
// The barrier proves the detached copy-back reached the tier GET and is
|
||||
// still paused, so the ongoing state and conflict rejection are not timing
|
||||
// assumptions about task scheduling.
|
||||
let ongoing = ecstore
|
||||
.get_object_info(bucket.as_str(), object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("Failed to load object info during restore");
|
||||
assert!(
|
||||
ongoing.restore_ongoing,
|
||||
"x-amz-restore must report ongoing-request=true right after the restore is accepted"
|
||||
"x-amz-restore must report ongoing-request=true while the copy-back tier GET is paused"
|
||||
);
|
||||
|
||||
// A second restore while one is in flight is rejected.
|
||||
@@ -2117,21 +2176,12 @@ async fn restore_object_usecase_reports_ongoing_conflict_and_completion() {
|
||||
"unexpected rejection for a repeated restore: {err:?}"
|
||||
);
|
||||
|
||||
get_barrier.release();
|
||||
|
||||
// Completion: ongoing flips to false and a future expiry-date appears.
|
||||
let mut completed = None;
|
||||
for _ in 0..40 {
|
||||
let info = ecstore
|
||||
.get_object_info(bucket.as_str(), object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("Failed to poll object info for restore completion");
|
||||
if !info.restore_ongoing && info.restore_expires.is_some() {
|
||||
completed = Some(info);
|
||||
break;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
let completed = completed.expect("restore copy-back should complete within the poll window");
|
||||
backend.clear_faults().await;
|
||||
let completed =
|
||||
wait_for_restore_completion(&ecstore, &backend, bucket.as_str(), object, RESTORE_COPY_BACK_WAIT_TIMEOUT).await;
|
||||
let completed = completed.unwrap_or_else(|err| panic!("{err}"));
|
||||
|
||||
let now_secs = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
@@ -2176,27 +2226,18 @@ async fn restore_object_usecase_accepts_exactly_one_of_two_concurrent_restores()
|
||||
let backend = register_mock_tier(&tier_name).await;
|
||||
|
||||
let bucket = format!("test-api-restore-cas-{}", &Uuid::new_v4().simple().to_string()[..8]);
|
||||
// Must live under the `test/` prefix — the shared transition rule filters on it.
|
||||
// Keep the object under the shared ILM test prefix for diagnostics parity.
|
||||
let object = "test/restore/cas-object.bin";
|
||||
let payload: Vec<u8> = (0..128 * 1024).map(|i| (i % 251) as u8).collect();
|
||||
|
||||
create_test_bucket(&ecstore, bucket.as_str()).await;
|
||||
set_bucket_lifecycle_transition_with_tier(bucket.as_str(), &tier_name)
|
||||
.await
|
||||
.expect("Failed to set lifecycle configuration");
|
||||
let _ = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
|
||||
let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
|
||||
let _ = transition_uploaded_object_directly(&ecstore, bucket.as_str(), object, &tier_name, &uploaded).await;
|
||||
backend.clear_op_log().await;
|
||||
|
||||
lifecycle::bucket_lifecycle_ops::enqueue_transition_for_existing_objects(ecstore.clone(), bucket.as_str())
|
||||
.await
|
||||
.expect("Failed to enqueue transitioned object");
|
||||
let _ = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
|
||||
.await
|
||||
.expect("object should transition before the concurrent restores run");
|
||||
|
||||
// Keep the winner's copy-back in flight while the loser's accept runs, so
|
||||
// the loser cannot slip into the already-restored path after a completed
|
||||
// copy-back.
|
||||
backend.set_latency(Some(Duration::from_millis(1500))).await;
|
||||
// Hold the accepted copy-back at the tier GET until both accept attempts
|
||||
// return, so the loser cannot observe an already-restored object.
|
||||
let get_barrier = backend.arm_get_barrier().await;
|
||||
|
||||
let tier_gets_before_restore = backend.get_count().await;
|
||||
|
||||
@@ -2241,27 +2282,34 @@ async fn restore_object_usecase_accepts_exactly_one_of_two_concurrent_restores()
|
||||
"the losing concurrent restore must be rejected as already in progress: {rejection:?}"
|
||||
);
|
||||
|
||||
// Let the single accepted copy-back complete, then verify the tier saw
|
||||
// exactly one restore read — a second GET means a double copy-back.
|
||||
let mut completed = false;
|
||||
for _ in 0..40 {
|
||||
let info = ecstore
|
||||
.get_object_info(bucket.as_str(), object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("Failed to poll object info for restore completion");
|
||||
if !info.restore_ongoing && info.restore_expires.is_some() {
|
||||
completed = true;
|
||||
get_barrier.wait_until_paused().await;
|
||||
get_barrier.release();
|
||||
|
||||
// This test is scoped to the accept CAS: completion, expiry metadata, and
|
||||
// local restored GET service are covered by the single-request restore test
|
||||
// above. Here it is enough to prove exactly one copy-back was admitted.
|
||||
let expected_tier_gets = tier_gets_before_restore + 1;
|
||||
let deadline = tokio::time::Instant::now() + TRANSITION_WAIT_TIMEOUT;
|
||||
loop {
|
||||
let actual_tier_gets = backend.get_count().await;
|
||||
if actual_tier_gets >= expected_tier_gets {
|
||||
assert_eq!(
|
||||
actual_tier_gets - tier_gets_before_restore,
|
||||
1,
|
||||
"two concurrent restore requests must trigger exactly one tier copy-back GET"
|
||||
);
|
||||
break;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
if tokio::time::Instant::now() >= deadline {
|
||||
let op_log = backend.op_log().await;
|
||||
panic!(
|
||||
"mock tier should record exactly one restore GET within {TRANSITION_WAIT_TIMEOUT:?}; \
|
||||
tier_gets_before_restore={tier_gets_before_restore}, actual_tier_gets={actual_tier_gets}, op_log={op_log:?}"
|
||||
);
|
||||
}
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
}
|
||||
backend.clear_faults().await;
|
||||
assert!(completed, "the accepted restore copy-back should complete within the poll window");
|
||||
assert_eq!(
|
||||
backend.get_count().await - tier_gets_before_restore,
|
||||
1,
|
||||
"two concurrent restore requests must trigger exactly one tier copy-back GET"
|
||||
);
|
||||
}
|
||||
|
||||
/// rustfs/backlog#1320: a single PUT must compute the replication decision
|
||||
|
||||
+17
-4
@@ -452,7 +452,7 @@ pub fn check_claims_from_token(token: &str, cred: &Credentials) -> S3Result<Hash
|
||||
return Err(s3_error!(InvalidRequest, "invalid token2"));
|
||||
}
|
||||
|
||||
if !cred.is_service_account() && cred.is_temp() && token != cred.session_token {
|
||||
if !cred.is_service_account() && cred.is_temp() && !constant_time_eq(token, &cred.session_token) {
|
||||
return Err(s3_error!(InvalidRequest, "invalid token3"));
|
||||
}
|
||||
|
||||
@@ -1801,9 +1801,10 @@ mod tests {
|
||||
#[test]
|
||||
fn test_constant_time_eq() {
|
||||
assert!(constant_time_eq("test", "test"));
|
||||
assert!(!constant_time_eq("test", "Test"));
|
||||
assert!(!constant_time_eq("test", "test1"));
|
||||
assert!(!constant_time_eq("test1", "test"));
|
||||
assert!(!constant_time_eq("Test", "test"), "first-byte mismatch must fail");
|
||||
assert!(!constant_time_eq("tesu", "test"), "last-byte mismatch must fail");
|
||||
assert!(!constant_time_eq("test", "test1"), "longer candidate must fail");
|
||||
assert!(!constant_time_eq("test1", "test"), "shorter candidate must fail");
|
||||
assert!(!constant_time_eq("", "test"));
|
||||
assert!(constant_time_eq("", ""));
|
||||
|
||||
@@ -1815,6 +1816,18 @@ mod tests {
|
||||
assert!(!constant_time_eq(key1, key3));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn session_token_comparison_uses_constant_time_helper() {
|
||||
let source = include_str!("auth.rs");
|
||||
let production = source.split_once("#[cfg(test)]").map_or(source, |(production, _)| production);
|
||||
let ordinary_comparison = ["token ", "!=", " cred.session_token"].concat();
|
||||
assert!(
|
||||
!production.contains(&ordinary_comparison),
|
||||
"temporary session tokens must not use ordinary string comparison"
|
||||
);
|
||||
assert!(production.contains("!constant_time_eq(token, &cred.session_token)"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_get_condition_values_source_ip() {
|
||||
let mut headers = HeaderMap::new();
|
||||
|
||||
+106
-18
@@ -14,7 +14,7 @@
|
||||
|
||||
use super::{
|
||||
module_switch::{resolve_notify_module_state, validate_notify_module_env, with_refreshed_notify_module_state_from},
|
||||
refresh_persisted_module_switches_from_store, runtime_sources,
|
||||
refresh_persisted_module_switches_from, runtime_sources,
|
||||
};
|
||||
use crate::storage_api::server::event::{
|
||||
EventArgs as EcstoreEventArgs, StorageObjectInfo, read_existing_server_config_no_lock, register_event_dispatch_hook,
|
||||
@@ -309,30 +309,36 @@ pub async fn shutdown_event_notifier() -> Result<(), NotificationError> {
|
||||
|
||||
#[instrument]
|
||||
pub async fn init_event_notifier() -> Result<(), NotificationError> {
|
||||
init_event_notifier_with_store(runtime_sources::current_object_store_handle).await
|
||||
}
|
||||
|
||||
async fn init_event_notifier_with_store<CurrentStore>(current_store: CurrentStore) -> Result<(), NotificationError>
|
||||
where
|
||||
CurrentStore: FnOnce() -> Option<std::sync::Arc<rustfs_notify::NotifyStore>>,
|
||||
{
|
||||
mark_event_notifier_unreconciled();
|
||||
validate_notify_module_env().map_err(NotificationError::Initialization)?;
|
||||
let system = ensure_live_events_initialized();
|
||||
refresh_persisted_module_switches_from_store()
|
||||
|
||||
let Some(store) = current_store() else {
|
||||
let enabled = refresh_notify_module_enabled();
|
||||
if enabled {
|
||||
return Err(NotificationError::Initialization(
|
||||
"failed to refresh notify module switch: storage layer not initialized".to_string(),
|
||||
));
|
||||
}
|
||||
initialize_live_event_support(&system, None).await?;
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
refresh_persisted_module_switches_from(store.clone())
|
||||
.await
|
||||
.map_err(|err| NotificationError::Initialization(format!("failed to refresh notify module switch: {err}")))?;
|
||||
|
||||
let enabled = refresh_notify_module_enabled();
|
||||
|
||||
if !enabled {
|
||||
info!(
|
||||
target: "rustfs::main::init_event_notifier",
|
||||
"Notify module is disabled, initializing live event stream support only. Set {}=true to enable notification targets.",
|
||||
rustfs_config::ENV_NOTIFY_ENABLE
|
||||
);
|
||||
if system.runtime_lifecycle_state() != NotificationRuntimeState::LiveOnly {
|
||||
system.set_targets_enabled(false, None).await?;
|
||||
}
|
||||
system.reload_persisted_config().await?;
|
||||
info!(
|
||||
target: "rustfs::main::init_event_notifier",
|
||||
"Live event stream support initialized successfully."
|
||||
);
|
||||
ensure_event_notifier_converged(&system)?;
|
||||
initialize_live_event_support(&system, Some(store)).await?;
|
||||
mark_event_notifier_reconciled();
|
||||
return Ok(());
|
||||
}
|
||||
@@ -347,7 +353,7 @@ pub async fn init_event_notifier() -> Result<(), NotificationError> {
|
||||
"Event notifier configuration found, proceeding with initialization."
|
||||
);
|
||||
|
||||
system.reload_persisted_config().await?;
|
||||
system.reload_persisted_config_from_store(store).await?;
|
||||
let runtime_state = system.runtime_lifecycle_state();
|
||||
if !matches!(runtime_state, NotificationRuntimeState::TargetsEnabled { .. }) {
|
||||
system.set_targets_enabled(true, None).await?;
|
||||
@@ -361,13 +367,53 @@ pub async fn init_event_notifier() -> Result<(), NotificationError> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn initialize_live_event_support(
|
||||
system: &NotificationSystem,
|
||||
store: Option<std::sync::Arc<rustfs_notify::NotifyStore>>,
|
||||
) -> Result<(), NotificationError> {
|
||||
if store.is_some() {
|
||||
info!(
|
||||
target: "rustfs::main::init_event_notifier",
|
||||
"Notify module is disabled, initializing live event stream support only. Set {}=true to enable notification targets.",
|
||||
rustfs_config::ENV_NOTIFY_ENABLE
|
||||
);
|
||||
} else {
|
||||
info!(
|
||||
target: "rustfs::main::init_event_notifier",
|
||||
"Notify module is disabled, initializing live event stream support only. Persisted notification target reconciliation is deferred until storage is initialized. Set {}=true to enable notification targets.",
|
||||
rustfs_config::ENV_NOTIFY_ENABLE
|
||||
);
|
||||
}
|
||||
|
||||
if system.runtime_lifecycle_state() != NotificationRuntimeState::LiveOnly {
|
||||
system.set_targets_enabled(false, None).await?;
|
||||
}
|
||||
if let Some(store) = store {
|
||||
system.reload_persisted_config_from_store(store).await?;
|
||||
}
|
||||
info!(
|
||||
target: "rustfs::main::init_event_notifier",
|
||||
"Live event stream support initialized successfully."
|
||||
);
|
||||
ensure_event_notifier_converged(system)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{convert_ecstore_object_info, parse_host_and_port, run_persisted_event_notifier_reconciler};
|
||||
use super::super::module_switch::{
|
||||
PersistedModuleSwitches, current_persisted_module_switches, persisted_module_switches_configured,
|
||||
set_persisted_module_switches,
|
||||
};
|
||||
use super::{
|
||||
convert_ecstore_object_info, init_event_notifier_with_store, parse_host_and_port, run_persisted_event_notifier_reconciler,
|
||||
};
|
||||
use crate::server::is_event_notifier_reconciled;
|
||||
use crate::storage_api::server::event::StorageObjectInfo;
|
||||
use crate::storage_api::server::event::contract::lifecycle::TransitionedObject;
|
||||
use chrono::{DateTime, Utc};
|
||||
use rustfs_notify::NotificationError;
|
||||
use rustfs_notify::NotificationRuntimeState;
|
||||
use serial_test::serial;
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
future::pending,
|
||||
@@ -381,6 +427,48 @@ mod tests {
|
||||
use tokio::sync::Notify;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn disabled_notify_without_storage_initializes_live_events_without_reconcile() {
|
||||
temp_env::async_with_vars([(rustfs_config::ENV_NOTIFY_ENABLE, Some("false"))], async {
|
||||
let previous = current_persisted_module_switches();
|
||||
let previous_configured = persisted_module_switches_configured();
|
||||
set_persisted_module_switches(PersistedModuleSwitches::default(), false);
|
||||
|
||||
init_event_notifier_with_store(|| None)
|
||||
.await
|
||||
.expect("disabled notify should not require storage during early bootstrap");
|
||||
|
||||
let system = rustfs_notify::notification_system().expect("live event container should be initialized");
|
||||
assert_eq!(system.runtime_lifecycle_state(), NotificationRuntimeState::LiveOnly);
|
||||
assert!(
|
||||
!is_event_notifier_reconciled(),
|
||||
"early live-only bootstrap must not claim persisted notify config is reconciled"
|
||||
);
|
||||
|
||||
set_persisted_module_switches(previous, previous_configured);
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn enabled_notify_without_storage_still_fails_visible() {
|
||||
temp_env::async_with_vars([(rustfs_config::ENV_NOTIFY_ENABLE, Some("true"))], async {
|
||||
let previous = current_persisted_module_switches();
|
||||
let previous_configured = persisted_module_switches_configured();
|
||||
set_persisted_module_switches(PersistedModuleSwitches::default(), false);
|
||||
|
||||
let err = init_event_notifier_with_store(|| None)
|
||||
.await
|
||||
.expect_err("enabled notify should still fail when storage is unavailable");
|
||||
assert!(err.to_string().contains("storage layer not initialized"));
|
||||
|
||||
set_persisted_module_switches(previous, previous_configured);
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_host_and_port_with_ipv4_and_port() {
|
||||
let (host, port) = parse_host_and_port("127.0.0.1:9000".to_string());
|
||||
|
||||
@@ -80,7 +80,7 @@ pub(crate) fn current_persisted_module_switches() -> PersistedModuleSwitches {
|
||||
}
|
||||
}
|
||||
|
||||
fn persisted_module_switches_configured() -> bool {
|
||||
pub(crate) fn persisted_module_switches_configured() -> bool {
|
||||
PERSISTED_MODULE_SWITCH_CONFIGURED.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user