mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-09 13:46:05 +00:00
chore(release): merge main into release
Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
@@ -197,6 +197,12 @@ test-group = 'e2e-cluster-nightly'
|
||||
filter = 'package(e2e_test) & (test(/^kms::kms_vault_test::/) | test(/^kms::kms_rekey_sweep_test::/) | test(/^kms::configured_roundtrip_test::test_configured_vault_kms_admin_and_versioned_cleanup$/))'
|
||||
test-group = 'e2e-vault'
|
||||
|
||||
# This four-disk, 65-member rollback probe already drives up to 32 concurrent
|
||||
# durable deletions. Reserve this nextest run's capacity for its progress oracle.
|
||||
[[profile.default.overrides]]
|
||||
filter = 'package(rustfs-ecstore) & test(=store::init::tests::dispatch_manifest_rollback_bounded_concurrency_reaches_tail_behind_slow_member)'
|
||||
threads-required = "num-test-threads"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ci profile — the strict CI gate (ci.yml `cargo nextest run --profile ci`)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -211,7 +211,10 @@ For developers who want to build RustFS Docker images from source with multi-arc
|
||||
|
||||
```bash
|
||||
# Build multi-architecture images locally
|
||||
./docker-buildx.sh --build-arg RELEASE=latest
|
||||
./docker-buildx.sh
|
||||
|
||||
# Build a single-platform image locally
|
||||
./docker-buildx.sh -p linux/amd64
|
||||
|
||||
# Build and push to registry
|
||||
./docker-buildx.sh --push
|
||||
|
||||
+4
-1
@@ -167,7 +167,10 @@ docker compose -f docker-compose-simple.yml up -d
|
||||
|
||||
```bash
|
||||
# 在本地构建多架构镜像
|
||||
./docker-buildx.sh --build-arg RELEASE=latest
|
||||
./docker-buildx.sh
|
||||
|
||||
# 在本地构建单平台镜像
|
||||
./docker-buildx.sh -p linux/amd64
|
||||
|
||||
# 构建并推送到仓库
|
||||
./docker-buildx.sh --push
|
||||
|
||||
@@ -1222,6 +1222,8 @@ mod tests {
|
||||
if let Some(erasure_set_drive_count) = evidence_case.and_then(|case| case.erasure_set_drive_count) {
|
||||
cluster.set_env("RUSTFS_ERASURE_SET_DRIVE_COUNT", erasure_set_drive_count.to_string());
|
||||
}
|
||||
// Capture physical baselines after the PUT rename fanout has drained.
|
||||
cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false");
|
||||
// Heal control uses the first lexicographically sorted grid host.
|
||||
// Keep that coordinator distinct from the remote target at index 1.
|
||||
cluster.nodes.sort_by(|left, right| left.url.cmp(&right.url));
|
||||
|
||||
@@ -143,6 +143,190 @@ async fn assert_degraded_cluster_publication_guard_errors_are_retryable() -> Tes
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn assert_quorum_object_body(client: &Client, bucket: &str, key: &str, expected: &[u8]) -> TestResult {
|
||||
let body = client
|
||||
.get_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.send()
|
||||
.await?
|
||||
.body
|
||||
.collect()
|
||||
.await?
|
||||
.into_bytes();
|
||||
assert_eq!(body.as_ref(), expected, "quorum read returned incorrect contents for {key}");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn wait_for_quorum_read_admission(clients: &[Client], bucket: &str) -> TestResult {
|
||||
// SIGKILL can orphan a granted lease. Wait for shared metadata-lock
|
||||
// admission before asserting the stable quorum boundary; cold bodies
|
||||
// remain unread throughout this readiness probe.
|
||||
let deadline =
|
||||
tokio::time::Instant::now() + rustfs_lock::fast_lock::DEFAULT_LOCK_TIMEOUT + std::time::Duration::from_secs(15);
|
||||
loop {
|
||||
let mut ready = true;
|
||||
for client in clients {
|
||||
for key in ["warm-small", "warm-large"] {
|
||||
match client.head_object().bucket(bucket).key(key).send().await {
|
||||
Ok(_) => {}
|
||||
Err(error) if error.raw_response().is_some_and(|response| response.status().as_u16() == 503) => {
|
||||
ready = false;
|
||||
break;
|
||||
}
|
||||
Err(error) => return Err(error.into()),
|
||||
}
|
||||
}
|
||||
if !ready {
|
||||
break;
|
||||
}
|
||||
}
|
||||
if ready {
|
||||
return Ok(());
|
||||
}
|
||||
if tokio::time::Instant::now() >= deadline {
|
||||
return Err(format!("read quorum did not become available after lease convergence for {bucket}").into());
|
||||
}
|
||||
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_degraded_cluster_read_quorum_follows_erasure_layout() -> TestResult {
|
||||
crate::common::init_logging();
|
||||
|
||||
for (node_count, parity) in [(4, 2), (6, 3), (6, 2)] {
|
||||
let read_quorum = node_count - parity;
|
||||
let write_quorum = read_quorum + usize::from(read_quorum == parity);
|
||||
let mut cluster = RustFSTestClusterEnvironment::new(node_count).await?;
|
||||
cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", format!("EC:{parity}"));
|
||||
// Wait for every seed fanout before removing any physical shard.
|
||||
cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false");
|
||||
cluster.set_env("RUSTFS_OBS_METRICS_EXPORT_ENABLED", "false");
|
||||
cluster.set_env("RUST_LOG", "warn,rustfs_lock=debug");
|
||||
cluster.start().await?;
|
||||
|
||||
let clients = cluster
|
||||
.create_all_clients()?
|
||||
.into_iter()
|
||||
.map(|client| {
|
||||
Client::from_conf(
|
||||
client
|
||||
.config()
|
||||
.to_builder()
|
||||
.retry_config(aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(1))
|
||||
.build(),
|
||||
)
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
let bucket = format!("read-quorum-{node_count}-{parity}");
|
||||
clients[0].create_bucket().bucket(&bucket).send().await?;
|
||||
let small = b"read quorum is derived from the erasure layout".to_vec();
|
||||
let large = (0..1_048_576)
|
||||
.map(|index| u8::try_from(index % 251).expect("bounded payload byte"))
|
||||
.collect::<Vec<_>>();
|
||||
for (key, body) in [
|
||||
("warm-small", &small),
|
||||
("warm-large", &large),
|
||||
("cold-small", &small),
|
||||
("cold-large", &large),
|
||||
("below-quorum", &large),
|
||||
] {
|
||||
clients[node_count - 1]
|
||||
.put_object()
|
||||
.bucket(&bucket)
|
||||
.key(key)
|
||||
.body(Bytes::copy_from_slice(body).into())
|
||||
.send()
|
||||
.await?;
|
||||
}
|
||||
for node in &cluster.nodes {
|
||||
for key in ["warm-small", "warm-large", "cold-small", "cold-large", "below-quorum"] {
|
||||
let census =
|
||||
crate::chaos::census_object_version_on_disk(std::path::Path::new(&node.data_dir), &bucket, key, None)?;
|
||||
assert!(census.is_complete(), "seed shard must be complete before fault injection: {census:?}");
|
||||
assert_eq!(census.data_blocks, Some(read_quorum));
|
||||
assert_eq!(census.parity_blocks, Some(parity));
|
||||
}
|
||||
}
|
||||
for client in &clients {
|
||||
assert_quorum_object_body(client, &bucket, "warm-small", &small).await?;
|
||||
assert_quorum_object_body(client, &bucket, "warm-large", &large).await?;
|
||||
}
|
||||
|
||||
for offline_node in (read_quorum..node_count).rev() {
|
||||
cluster.stop_node(offline_node)?;
|
||||
wait_for_quorum_read_admission(&clients[..offline_node], &bucket).await?;
|
||||
for client in clients.iter().take(offline_node) {
|
||||
client.head_bucket().bucket(&bucket).send().await?;
|
||||
assert_quorum_object_body(client, &bucket, "warm-large", &large).await?;
|
||||
}
|
||||
}
|
||||
|
||||
// Exercise more than the five-second positive bucket-validation TTL.
|
||||
// Every sample must succeed; polling must not hide a transient failure.
|
||||
let validation_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(6);
|
||||
loop {
|
||||
for client in clients.iter().take(read_quorum) {
|
||||
assert_quorum_object_body(client, &bucket, "warm-small", &small).await?;
|
||||
assert_quorum_object_body(client, &bucket, "warm-large", &large).await?;
|
||||
let listing = client.list_objects_v2().bucket(&bucket).send().await?;
|
||||
for key in ["warm-small", "warm-large", "cold-small", "cold-large", "below-quorum"] {
|
||||
assert!(listing.contents().iter().any(|entry| entry.key() == Some(key)), "listing omitted {key}");
|
||||
}
|
||||
}
|
||||
if tokio::time::Instant::now() >= validation_deadline {
|
||||
break;
|
||||
}
|
||||
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
|
||||
}
|
||||
for client in clients.iter().take(read_quorum) {
|
||||
assert_quorum_object_body(client, &bucket, "cold-small", &small).await?;
|
||||
assert_quorum_object_body(client, &bucket, "cold-large", &large).await?;
|
||||
}
|
||||
|
||||
let write = clients[0]
|
||||
.put_object()
|
||||
.bucket(&bucket)
|
||||
.key("quorum-write")
|
||||
.body(Bytes::copy_from_slice(&small).into())
|
||||
.send()
|
||||
.await;
|
||||
if read_quorum >= write_quorum {
|
||||
write?;
|
||||
} else {
|
||||
let error = write.expect_err("a read quorum must not authorize a write that needs more votes");
|
||||
assert_eq!(error.as_service_error().and_then(|error| error.meta().code()), Some("ServiceUnavailable"));
|
||||
}
|
||||
|
||||
cluster.stop_node(read_quorum - 1)?;
|
||||
for client in clients.iter().take(read_quorum - 1) {
|
||||
match client.get_object().bucket(&bucket).key("below-quorum").send().await {
|
||||
Ok(response) => assert!(
|
||||
response.body.collect().await.is_err(),
|
||||
"fewer than {read_quorum} valid fragments must not reconstruct an uncached object"
|
||||
),
|
||||
Err(error) => assert_eq!(
|
||||
error.as_service_error().and_then(|error| error.meta().code()),
|
||||
Some("ServiceUnavailable"),
|
||||
"a quorum loss must not be mistaken for a missing object"
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
for node in 0..read_quorum - 1 {
|
||||
cluster.stop_node(node)?;
|
||||
}
|
||||
cluster.start().await?;
|
||||
for client in &clients {
|
||||
assert_quorum_object_body(client, &bucket, "warm-large", &large).await?;
|
||||
assert_quorum_object_body(client, &bucket, "below-quorum", &large).await?;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn put_object(client: Client, payload: Vec<u8>, writer_id: usize) -> Result<(), String> {
|
||||
client
|
||||
.put_object()
|
||||
|
||||
@@ -627,7 +627,13 @@ impl From<tokio::task::JoinError> for DiskError {
|
||||
impl Clone for DiskError {
|
||||
fn clone(&self) -> Self {
|
||||
match self {
|
||||
DiskError::Io(io_error) => DiskError::Io(std::io::Error::new(io_error.kind(), io_error.to_string())),
|
||||
DiskError::Io(io_error) => DiskError::Io(
|
||||
rustfs_rio::clone_internode_http_io_error(io_error)
|
||||
.and_then(std::io::Error::into_inner)
|
||||
// The helper derives a kind from the source; Clone must retain the original outer kind.
|
||||
.map(|source| std::io::Error::new(io_error.kind(), source))
|
||||
.unwrap_or_else(|| std::io::Error::new(io_error.kind(), io_error.to_string())),
|
||||
),
|
||||
DiskError::MaxVersionsExceeded => DiskError::MaxVersionsExceeded,
|
||||
DiskError::Unexpected => DiskError::Unexpected,
|
||||
DiskError::CorruptedFormat => DiskError::CorruptedFormat,
|
||||
@@ -1265,6 +1271,49 @@ mod tests {
|
||||
assert!(!bad_request.is_retryable_internode_write_failure());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_internode_http_clone_preserves_retryability_status_and_context() {
|
||||
use http::StatusCode;
|
||||
use rustfs_rio::InternodeHttpErrorKind::{ConnectionRefused, ConnectionReset, HttpStatus, Unknown};
|
||||
|
||||
for (kind, retryable) in [
|
||||
(ConnectionRefused, true),
|
||||
(ConnectionReset, true),
|
||||
(HttpStatus(StatusCode::TOO_MANY_REQUESTS), true),
|
||||
(HttpStatus(StatusCode::SERVICE_UNAVAILABLE), true),
|
||||
(HttpStatus(StatusCode::CONFLICT), true),
|
||||
(Unknown, false),
|
||||
(HttpStatus(StatusCode::BAD_REQUEST), false),
|
||||
(HttpStatus(StatusCode::INTERNAL_SERVER_ERROR), false),
|
||||
] {
|
||||
let original = DiskError::from(rustfs_rio::new_test_internode_http_io_error(kind));
|
||||
assert_eq!(original.internode_http_error_kind(), Some(kind));
|
||||
assert_eq!(original.is_retryable_internode_write_failure(), retryable);
|
||||
|
||||
let cloned = original.clone();
|
||||
assert_eq!(cloned, original, "clone must preserve the error bucket for {kind:?}");
|
||||
assert_eq!(
|
||||
cloned.is_retryable_internode_write_failure(),
|
||||
retryable,
|
||||
"clone changed retryability for {kind:?}"
|
||||
);
|
||||
assert_eq!(cloned.internode_http_error_kind(), Some(kind));
|
||||
if let HttpStatus(status) = kind {
|
||||
assert!(cloned.is_internode_http_status(status.as_u16()));
|
||||
}
|
||||
let DiskError::Io(io_error) = &cloned else {
|
||||
panic!("unmarked internode error must remain Io: {cloned:?}");
|
||||
};
|
||||
let source = io_error
|
||||
.get_ref()
|
||||
.and_then(|source| source.downcast_ref::<InternodeHttpError>())
|
||||
.expect("clone must retain the structured internode error");
|
||||
assert_eq!(source.context().method(), "PUT");
|
||||
assert_eq!(source.context().target(), "/rustfs/rpc/put_file_stream");
|
||||
assert_eq!(source.context().operation(), Some(INTERNODE_OPERATION_PUT_FILE_STREAM));
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn read_stream_conflict_is_not_a_retryable_put_file_failure() {
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
@@ -1309,11 +1358,57 @@ mod tests {
|
||||
!error.is_retryable_internode_write_failure(),
|
||||
"read-operation 409 must not trigger put-file retry"
|
||||
);
|
||||
let cloned = error.clone();
|
||||
let reduced = crate::disk::error_reduce::reduce_write_quorum_errs(&[Some(error)], &[], 1)
|
||||
.expect("the read conflict must remain the dominant error");
|
||||
for preserved in [&cloned, &reduced] {
|
||||
assert!(
|
||||
!preserved.is_retryable_internode_write_failure(),
|
||||
"cloning or reducing a read conflict must not turn it into a PUT retry"
|
||||
);
|
||||
assert!(preserved.is_internode_http_status(409));
|
||||
let DiskError::Io(io_error) = preserved else {
|
||||
panic!("read conflict must remain Io: {preserved:?}");
|
||||
};
|
||||
let source = io_error
|
||||
.get_ref()
|
||||
.and_then(|source| source.downcast_ref::<InternodeHttpError>())
|
||||
.expect("read conflict must retain its request context");
|
||||
assert_eq!(source.context().method(), "GET");
|
||||
assert_eq!(source.context().target(), "/rustfs/rpc/read_file_stream");
|
||||
assert_eq!(
|
||||
source.context().operation(),
|
||||
Some(rustfs_io_metrics::internode_metrics::INTERNODE_OPERATION_READ_FILE_STREAM)
|
||||
);
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("isolated read-conflict test must finish within its budget");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_internode_http_clone_preserves_outer_io_kind_and_message() {
|
||||
let source = rustfs_rio::new_test_internode_http_io_error(InternodeHttpErrorKind::ConnectionReset)
|
||||
.into_inner()
|
||||
.expect("the internode helper must provide a typed source");
|
||||
let original_io = io::Error::new(io::ErrorKind::InvalidData, source);
|
||||
let message = original_io.to_string();
|
||||
let original = DiskError::from(original_io);
|
||||
assert_eq!(original.internode_http_error_kind(), Some(InternodeHttpErrorKind::ConnectionReset));
|
||||
assert!(original.is_retryable_internode_write_failure());
|
||||
|
||||
let cloned = original.clone();
|
||||
let reduced = crate::disk::error_reduce::reduce_write_quorum_errs(&[Some(original)], &[], 1)
|
||||
.expect("the wrapped internode error must remain the dominant error");
|
||||
for preserved in [&cloned, &reduced] {
|
||||
let DiskError::Io(io_error) = preserved else {
|
||||
panic!("the wrapped error must remain Io: {preserved:?}");
|
||||
};
|
||||
assert_eq!(io_error.kind(), io::ErrorKind::InvalidData);
|
||||
assert_eq!(io_error.to_string(), message);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_internode_missing_errors_preserve_disk_error_types() {
|
||||
let file_missing = DiskError::from(rustfs_rio::new_test_remote_file_not_found_http_io_error());
|
||||
@@ -1325,6 +1420,17 @@ mod tests {
|
||||
assert_eq!(file_missing, DiskError::FileNotFound);
|
||||
assert_eq!(volume_missing, DiskError::VolumeNotFound);
|
||||
assert!(matches!(unmarked_server_error, DiskError::Io(_)));
|
||||
for missing in [file_missing, volume_missing] {
|
||||
assert_eq!(missing.clone(), missing);
|
||||
assert_eq!(
|
||||
crate::disk::error_reduce::reduce_write_quorum_errs(
|
||||
&[Some(missing.clone()), Some(missing.clone()), None],
|
||||
&[],
|
||||
2
|
||||
),
|
||||
Some(missing)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -226,6 +226,78 @@ mod tests {
|
||||
assert_eq!(res, Some(quorum_err));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_write_quorum_reduction_preserves_internode_http_identity() {
|
||||
use http::StatusCode;
|
||||
use rustfs_rio::InternodeHttpErrorKind::{ConnectionRefused, HttpStatus, Unknown};
|
||||
|
||||
for (kind, retryable) in [
|
||||
(ConnectionRefused, true),
|
||||
(HttpStatus(StatusCode::SERVICE_UNAVAILABLE), true),
|
||||
(HttpStatus(StatusCode::CONFLICT), true),
|
||||
(Unknown, false),
|
||||
(HttpStatus(StatusCode::BAD_REQUEST), false),
|
||||
] {
|
||||
// Construct both producer errors independently: the reducer owns the first clone.
|
||||
let first = Error::from(rustfs_rio::new_test_internode_http_io_error(kind));
|
||||
let second = Error::from(rustfs_rio::new_test_internode_http_io_error(kind));
|
||||
assert_eq!(first.internode_http_error_kind(), Some(kind));
|
||||
assert_eq!(second.internode_http_error_kind(), Some(kind));
|
||||
assert_eq!(first.is_retryable_internode_write_failure(), retryable);
|
||||
let errors = [Some(first), Some(second), None];
|
||||
let reduced = reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, 2)
|
||||
.expect("two equal producer errors must dominate one successful write");
|
||||
|
||||
assert_eq!(Some(&reduced), errors[0].as_ref());
|
||||
assert_eq!(
|
||||
reduced.is_retryable_internode_write_failure(),
|
||||
retryable,
|
||||
"quorum reduction changed retryability for {kind:?}"
|
||||
);
|
||||
assert_eq!(reduced.internode_http_error_kind(), Some(kind));
|
||||
if let HttpStatus(status) = kind {
|
||||
assert!(reduced.is_internode_http_status(status.as_u16()));
|
||||
}
|
||||
let Error::Io(io_error) = &reduced else {
|
||||
panic!("the dominant error must remain Io: {reduced:?}");
|
||||
};
|
||||
let source = io_error
|
||||
.get_ref()
|
||||
.and_then(|source| source.downcast_ref::<rustfs_rio::InternodeHttpError>())
|
||||
.expect("quorum reduction must retain the structured internode error");
|
||||
assert_eq!(source.context().method(), "PUT");
|
||||
assert_eq!(source.context().target(), "/rustfs/rpc/put_file_stream");
|
||||
assert_eq!(
|
||||
source.context().operation(),
|
||||
Some(rustfs_io_metrics::internode_metrics::INTERNODE_OPERATION_PUT_FILE_STREAM)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_clone_and_write_quorum_do_not_promote_non_retryable_errors() {
|
||||
use http::StatusCode;
|
||||
use rustfs_rio::InternodeHttpErrorKind::{HttpStatus, Unknown};
|
||||
|
||||
for original in [
|
||||
Error::from(rustfs_rio::new_test_internode_http_io_error(Unknown)),
|
||||
Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(StatusCode::BAD_REQUEST))),
|
||||
Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(StatusCode::FORBIDDEN))),
|
||||
Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(StatusCode::NOT_FOUND))),
|
||||
Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
))),
|
||||
err_io("internode connection reset: PUT /rustfs/rpc/put_file_stream"),
|
||||
] {
|
||||
assert!(!original.is_retryable_internode_write_failure());
|
||||
let cloned = original.clone();
|
||||
let reduced =
|
||||
reduce_write_quorum_errs(&[Some(original)], &[], 1).expect("a non-retryable error must remain an error");
|
||||
assert!(!cloned.is_retryable_internode_write_failure());
|
||||
assert!(!reduced.is_retryable_internode_write_failure());
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_count_errs() {
|
||||
let e1 = err_io("a");
|
||||
|
||||
@@ -326,14 +326,15 @@ impl SetDisks {
|
||||
let parity_blocks = Self::common_parity(&parities, default_parity_count as i32);
|
||||
|
||||
if parity_blocks < 0 {
|
||||
// No parity value reached read quorum. Distinguish two cases:
|
||||
// enough disks answered with valid-looking metadata that simply
|
||||
// cannot be reconciled (corrupt/foreign entries — retrying cannot
|
||||
// help, and heal should see Corrupt, rustfs#5801) versus too few
|
||||
// healthy answers (a genuine quorum condition where retry may
|
||||
// succeed once disks recover).
|
||||
// A consistent layout can require more replies than the initial
|
||||
// half-set probe. Reaching that probe alone is not corruption;
|
||||
// only invalid or conflicting healthy replies establish that.
|
||||
let healthy_replies = errs.iter().filter(|err| err.is_none()).count();
|
||||
if healthy_replies >= expected_rquorum {
|
||||
let consistent_parity = parities
|
||||
.iter()
|
||||
.find(|&&parity| parity >= 0)
|
||||
.filter(|&&parity| parities.iter().filter(|&&candidate| candidate == parity).count() == healthy_replies);
|
||||
if healthy_replies >= expected_rquorum && consistent_parity.is_none() {
|
||||
error!(
|
||||
"object_quorum_from_meta: irreconcilable parity across {healthy_replies} healthy replies (corrupt metadata), errs={errs:?}"
|
||||
);
|
||||
@@ -1652,6 +1653,40 @@ mod tests {
|
||||
assert_eq!(err, DiskError::FileCorrupt);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn consistent_parity_below_its_data_shard_quorum_is_not_corruption() {
|
||||
for (drive_count, parity) in [(6, 2), (8, 2), (12, 4)] {
|
||||
let data = drive_count - parity;
|
||||
let mut metas = (1..=drive_count)
|
||||
.map(|index| {
|
||||
let mut info = FileInfo::new("bucket/object", data, parity);
|
||||
info.size = 1024;
|
||||
info.erasure.index = index;
|
||||
info
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
let mut errs = vec![Some(DiskError::DiskNotFound); drive_count];
|
||||
errs[..data].fill(None);
|
||||
assert_eq!(
|
||||
SetDisks::object_quorum_from_meta(&metas, &errs, parity).expect("exact data quorum should resolve"),
|
||||
(data as i32, data as i32)
|
||||
);
|
||||
|
||||
errs[data - 1] = Some(DiskError::DiskNotFound);
|
||||
assert_eq!(
|
||||
SetDisks::object_quorum_from_meta(&metas, &errs, parity).expect_err("one fewer shard cannot resolve"),
|
||||
DiskError::ErasureReadQuorum,
|
||||
"layout {drive_count}/{parity} has consistent metadata but insufficient shards"
|
||||
);
|
||||
|
||||
metas[0].erasure.parity_blocks = usize::MAX;
|
||||
assert_eq!(
|
||||
SetDisks::object_quorum_from_meta(&metas, &errs, parity).expect_err("corrupt healthy replies must be rejected"),
|
||||
DiskError::FileCorrupt
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Too few healthy replies remains a genuine quorum condition where a
|
||||
/// retry may succeed once disks recover.
|
||||
#[test]
|
||||
|
||||
@@ -865,6 +865,7 @@ pub(crate) use core::io_primitives::{ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, ren
|
||||
mod ctx;
|
||||
mod metadata;
|
||||
mod ops;
|
||||
pub(crate) use ops::bucket::BucketInfoQuorum;
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) use ops::hermetic_set_disks_isolated;
|
||||
|
||||
@@ -21,12 +21,72 @@
|
||||
|
||||
use super::super::{
|
||||
BUCKET_OP_IGNORED_ERRS, BucketInfo, BucketOperations, BucketOptions, DeleteBucketOptions, DiskError, Error, HashMap,
|
||||
MakeBucketOptions, Result, SetDisks, is_reserved_or_invalid_bucket, join_all, reduce_write_quorum_errs,
|
||||
MakeBucketOptions, Result, SetDisks, is_reserved_or_invalid_bucket, join_all, reduce_read_quorum_errs,
|
||||
reduce_write_quorum_errs,
|
||||
};
|
||||
use crate::api::bucket::metadata_sys;
|
||||
use crate::disk::DiskAPI;
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
pub(crate) enum BucketInfoQuorum {
|
||||
Read,
|
||||
Write,
|
||||
}
|
||||
|
||||
impl SetDisks {
|
||||
pub(crate) async fn stat_bucket_with_quorum(&self, bucket: &str, quorum: BucketInfoQuorum) -> Result<BucketInfo> {
|
||||
let disks = self.disk_inventory().await;
|
||||
let disk_count = disks.len();
|
||||
let mut futures = Vec::with_capacity(disk_count);
|
||||
for disk in disks {
|
||||
let bucket = bucket.to_string();
|
||||
futures.push(async move {
|
||||
match disk {
|
||||
Some(disk) => disk.stat_volume(&bucket).await,
|
||||
None => Err(DiskError::DiskNotFound),
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
let results = join_all(futures).await;
|
||||
let mut infos = Vec::with_capacity(results.len());
|
||||
let mut errs = Vec::with_capacity(results.len());
|
||||
for result in results {
|
||||
match result {
|
||||
Ok(info) => {
|
||||
infos.push(Some(info));
|
||||
errs.push(None);
|
||||
}
|
||||
Err(err) => {
|
||||
infos.push(None);
|
||||
errs.push(Some(err));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let error = match quorum {
|
||||
// Bucket mutations use a majority regardless of object storage
|
||||
// class. A namespace read must intersect that majority; object
|
||||
// readers still enforce the persisted layout's data-shard quorum.
|
||||
BucketInfoQuorum::Read => reduce_read_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, disk_count.div_ceil(2).max(1)),
|
||||
BucketInfoQuorum::Write => reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, disk_count / 2 + 1),
|
||||
};
|
||||
if let Some(err) = error {
|
||||
return Err(err.into());
|
||||
}
|
||||
|
||||
infos
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.next()
|
||||
.map(|info| BucketInfo {
|
||||
name: info.name,
|
||||
created: info.created,
|
||||
..Default::default()
|
||||
})
|
||||
.ok_or(Error::VolumeNotFound)
|
||||
}
|
||||
|
||||
pub(crate) async fn list_bucket_for_scanner(&self, _opts: &BucketOptions) -> Result<(Vec<BucketInfo>, bool)> {
|
||||
let disks = self.disk_inventory().await;
|
||||
let write_quorum = (disks.len() / 2) + 1;
|
||||
@@ -131,59 +191,12 @@ impl BucketOperations for SetDisks {
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn get_bucket_info(&self, bucket: &str, _opts: &BucketOptions) -> Result<BucketInfo> {
|
||||
let disks = self.disk_inventory().await;
|
||||
let write_quorum = (disks.len() / 2) + 1;
|
||||
|
||||
let mut futures = Vec::with_capacity(disks.len());
|
||||
for disk in disks {
|
||||
let bucket = bucket.to_string();
|
||||
futures.push(async move {
|
||||
match disk {
|
||||
Some(disk) => disk.stat_volume(&bucket).await,
|
||||
None => Err(DiskError::DiskNotFound),
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
let results = join_all(futures).await;
|
||||
let mut infos = Vec::with_capacity(results.len());
|
||||
let mut errs = Vec::with_capacity(results.len());
|
||||
for result in results {
|
||||
match result {
|
||||
Ok(info) => {
|
||||
infos.push(Some(info));
|
||||
errs.push(None);
|
||||
}
|
||||
Err(err) => {
|
||||
infos.push(None);
|
||||
errs.push(Some(err));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) {
|
||||
return Err(err.into());
|
||||
}
|
||||
|
||||
let mut versioning = false;
|
||||
let mut object_locking = false;
|
||||
let mut info = self.stat_bucket_with_quorum(bucket, BucketInfoQuorum::Write).await?;
|
||||
if let Ok(sys) = metadata_sys::get(bucket).await {
|
||||
versioning = sys.versioning();
|
||||
object_locking = sys.object_locking();
|
||||
info.versioning = sys.versioning();
|
||||
info.object_locking = sys.object_locking();
|
||||
}
|
||||
|
||||
infos
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.next()
|
||||
.map(|info| BucketInfo {
|
||||
name: info.name,
|
||||
created: info.created,
|
||||
versioning,
|
||||
object_locking,
|
||||
..Default::default()
|
||||
})
|
||||
.ok_or(Error::VolumeNotFound)
|
||||
Ok(info)
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
|
||||
@@ -19,7 +19,7 @@ use crate::bucket::{
|
||||
};
|
||||
use crate::error::is_err_bucket_not_found;
|
||||
use crate::runtime::sources as runtime_sources;
|
||||
use crate::set_disk::get_lock_acquire_timeout;
|
||||
use crate::set_disk::{BucketInfoQuorum, get_lock_acquire_timeout};
|
||||
use crate::storage_api_contracts::bucket::{BUCKET_LIFECYCLE_LOCK_OBJECT, SRBucketDeleteOp};
|
||||
use crate::storage_api_contracts::namespace::NamespaceLocking as _;
|
||||
use futures::stream::{self, StreamExt};
|
||||
@@ -797,17 +797,30 @@ impl ECStore {
|
||||
|
||||
#[instrument(skip(self))]
|
||||
pub(crate) async fn get_bucket_info_from_sets(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
|
||||
self.get_bucket_info_from_sets_with_quorum(bucket, opts, BucketInfoQuorum::Write)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn get_bucket_info_from_sets_with_quorum(
|
||||
&self,
|
||||
bucket: &str,
|
||||
opts: &BucketOptions,
|
||||
quorum: BucketInfoQuorum,
|
||||
) -> Result<BucketInfo> {
|
||||
// One host may participate in several pools after expansion. Resolve the
|
||||
// namespace against each erasure set so disks from different pools can
|
||||
// never be combined into one bucket quorum.
|
||||
// Bucket validation is request-path IO. Keep the previous peer fanout's
|
||||
// latency shape by probing every set concurrently; scanner listings use
|
||||
// a separate bounded path below because they run continuously.
|
||||
let mut scoped_results =
|
||||
futures::future::join_all(self.bucket_sets().map(|(pool_index, set_index, set)| async move {
|
||||
(pool_index, set_index, set.get_bucket_info(bucket, opts).await)
|
||||
}))
|
||||
.await;
|
||||
let mut scoped_results = futures::future::join_all(self.bucket_sets().map(|(pool_index, set_index, set)| async move {
|
||||
let result = match quorum {
|
||||
BucketInfoQuorum::Read => set.stat_bucket_with_quorum(bucket, quorum).await,
|
||||
BucketInfoQuorum::Write => set.get_bucket_info(bucket, opts).await,
|
||||
};
|
||||
(pool_index, set_index, result)
|
||||
}))
|
||||
.await;
|
||||
scoped_results.sort_unstable_by_key(|(pool_index, set_index, _)| (*pool_index, *set_index));
|
||||
|
||||
let mut first_info = None;
|
||||
@@ -831,7 +844,11 @@ impl ECStore {
|
||||
|
||||
#[instrument(skip(self))]
|
||||
pub(super) async fn handle_get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
|
||||
let mut info = self.get_bucket_info_from_sets(bucket, opts).await?;
|
||||
let mut info = match self.get_bucket_info_from_sets(bucket, opts).await {
|
||||
Ok(info) => info,
|
||||
Err(Error::ErasureWriteQuorum) => return self.get_bucket_info_at_read_quorum(bucket, opts).await,
|
||||
Err(err) => return Err(err),
|
||||
};
|
||||
|
||||
if let Ok(sys) = metadata_sys::get_in(&self.ctx, bucket).await {
|
||||
if should_override_created_from_metadata(sys.created) {
|
||||
@@ -844,6 +861,35 @@ impl ECStore {
|
||||
Ok(info)
|
||||
}
|
||||
|
||||
async fn get_bucket_info_at_read_quorum(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
|
||||
// Lock order: bucket lifecycle -> internal metadata object read locks.
|
||||
// Keep create/delete from changing the namespace while a read quorum
|
||||
// confirms both physical presence and persisted bucket metadata.
|
||||
let guard = self.acquire_bucket_lifecycle_read_lock(bucket).await?;
|
||||
await_bucket_namespace_operation(Some(&guard), bucket, "bucket read quorum validation", async {
|
||||
let mut info = self
|
||||
.get_bucket_info_from_sets_with_quorum(bucket, opts, BucketInfoQuorum::Read)
|
||||
.await?;
|
||||
let (metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?;
|
||||
if !persisted {
|
||||
// A minority of directories left by failed creation is not an
|
||||
// authoritative bucket. Never turn fabricated defaults into
|
||||
// permission to serve degraded reads.
|
||||
return Err(Error::ErasureReadQuorum);
|
||||
}
|
||||
if metadata.name != bucket {
|
||||
return Err(Error::FileCorrupt);
|
||||
}
|
||||
if should_override_created_from_metadata(metadata.created) {
|
||||
info.created = Some(metadata.created);
|
||||
}
|
||||
info.versioning = metadata.versioning();
|
||||
info.object_locking = metadata.object_locking();
|
||||
Ok(info)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
#[instrument(skip(self))]
|
||||
pub(super) async fn handle_list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
|
||||
// TODO(backlog): support cached bucket listing via opts.cached
|
||||
@@ -1074,7 +1120,7 @@ mod tests {
|
||||
run_physical_bucket_deletion, scan_metadata_less_residue, scan_metadata_less_residue_with_budget,
|
||||
should_override_created_from_metadata, validate_table_bucket_delete_allowed,
|
||||
};
|
||||
use crate::bucket::metadata::table_bucket_catalog_metadata_prefix;
|
||||
use crate::bucket::metadata::{BucketMetadata, table_bucket_catalog_metadata_prefix};
|
||||
use crate::bucket::metadata_sys;
|
||||
use crate::cluster::rpc::peer_s3_client::install_delete_bucket_empty_scan_barrier;
|
||||
use crate::disk::{BUCKET_META_PREFIX, DiskAPI, RUSTFS_META_BUCKET, STORAGE_FORMAT_FILE};
|
||||
@@ -1101,6 +1147,7 @@ mod tests {
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::{Duration, SystemTime};
|
||||
use time::OffsetDateTime;
|
||||
use tokio::io::AsyncReadExt;
|
||||
use tokio::sync::{Notify, OnceCell};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use uuid::Uuid;
|
||||
@@ -1384,11 +1431,18 @@ mod tests {
|
||||
}
|
||||
|
||||
async fn setup_multi_pool_bucket_test_env() -> (tempfile::TempDir, Arc<ECStore>) {
|
||||
setup_bucket_quorum_test_env(&[4, 4], None).await
|
||||
}
|
||||
|
||||
async fn setup_bucket_quorum_test_env(
|
||||
drives_per_pool: &[usize],
|
||||
standard_parity: Option<usize>,
|
||||
) -> (tempfile::TempDir, Arc<ECStore>) {
|
||||
let temp_dir = tempfile::tempdir().expect("multi-pool bucket test directory should be created");
|
||||
let mut pools = Vec::new();
|
||||
for pool_index in 0..2 {
|
||||
for (pool_index, &drive_count) in drives_per_pool.iter().enumerate() {
|
||||
let mut endpoints = Vec::new();
|
||||
for disk_index in 0..4 {
|
||||
for disk_index in 0..drive_count {
|
||||
let disk_path = temp_dir.path().join(format!("pool{pool_index}-disk{disk_index}"));
|
||||
tokio::fs::create_dir_all(&disk_path)
|
||||
.await
|
||||
@@ -1403,7 +1457,7 @@ mod tests {
|
||||
pools.push(PoolEndpoints {
|
||||
legacy: false,
|
||||
set_count: 1,
|
||||
drives_per_set: 4,
|
||||
drives_per_set: drive_count,
|
||||
endpoints: Endpoints::from(endpoints),
|
||||
cmd_line: format!("bucket-test-pool-{pool_index}"),
|
||||
platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
|
||||
@@ -1424,9 +1478,12 @@ mod tests {
|
||||
)
|
||||
.await
|
||||
.expect("multi-pool ECStore should initialize");
|
||||
let storage_class =
|
||||
crate::config::storageclass::lookup_config_for_pools_without_env(&rustfs_config::server_config::KVS::new(), &[4, 4])
|
||||
.expect("multi-pool storage class should match both four-disk pools");
|
||||
let mut storage_class_kvs = rustfs_config::server_config::KVS::new();
|
||||
if let Some(parity) = standard_parity {
|
||||
storage_class_kvs.insert(crate::config::storageclass::CLASS_STANDARD.to_string(), format!("EC:{parity}"));
|
||||
}
|
||||
let storage_class = crate::config::storageclass::lookup_config_for_pools_without_env(&storage_class_kvs, drives_per_pool)
|
||||
.expect("storage class should match every test erasure set");
|
||||
for pool in &ecstore.pools {
|
||||
for set in &pool.disk_set {
|
||||
set.set_test_storage_class_config(storage_class.clone());
|
||||
@@ -2184,6 +2241,218 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn bucket_info_read_quorum_tracks_erasure_layout() {
|
||||
for (drive_count, parity) in [(2, 1), (3, 1), (4, 2), (5, 2), (6, 3), (8, 4), (6, 2), (12, 6)] {
|
||||
let (_temp_dir, store) = setup_bucket_quorum_test_env(&[drive_count], Some(parity)).await;
|
||||
metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
|
||||
let bucket = format!("read-quorum-{drive_count}-{parity}");
|
||||
let object = "uncached-object";
|
||||
let body = b"erasure read quorum must follow the persisted layout".repeat(32_768);
|
||||
store
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("healthy namespace should accept bucket creation");
|
||||
store
|
||||
.put_object(&bucket, object, &mut PutObjReader::from_vec(body.clone()), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("healthy erasure set should accept the seed object");
|
||||
let set = &store.pools[0].disk_set[0];
|
||||
let lock = set
|
||||
.new_ns_lock(&bucket, object)
|
||||
.await
|
||||
.expect("seed namespace lock should resolve");
|
||||
drop(
|
||||
lock.get_write_lock(Duration::from_secs(30))
|
||||
.await
|
||||
.expect("seed physical fanout must finish before taking disks offline"),
|
||||
);
|
||||
if (drive_count, parity) == (6, 3) {
|
||||
let mut kvs = rustfs_config::server_config::KVS::new();
|
||||
kvs.insert(crate::config::storageclass::CLASS_STANDARD.to_string(), "EC:2".to_string());
|
||||
set.set_test_storage_class_config(
|
||||
crate::config::storageclass::lookup_config_for_pools_without_env(&kvs, &[drive_count])
|
||||
.expect("a later storage-class change must not raise old objects' read quorum"),
|
||||
);
|
||||
}
|
||||
|
||||
let offline_indexes = (0..parity).collect::<Vec<_>>();
|
||||
let offline = take_set_disks_offline(&store, set, &offline_indexes).await;
|
||||
let info = store
|
||||
.get_bucket_info(&bucket, &BucketOptions::default())
|
||||
.await
|
||||
.expect("bucket validation must admit the object's exact read quorum");
|
||||
assert_eq!(info.name, bucket);
|
||||
|
||||
let mut reader = store
|
||||
.get_object_reader(&bucket, object, None, Default::default(), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("the persisted layout should remain readable at its exact data-shard quorum");
|
||||
let mut restored = Vec::new();
|
||||
reader
|
||||
.stream
|
||||
.read_to_end(&mut restored)
|
||||
.await
|
||||
.expect("quorum read should reconstruct the body");
|
||||
assert_eq!(restored, body, "layout {drive_count}/{parity} must retain exact object contents");
|
||||
drop(reader);
|
||||
|
||||
if drive_count - parity == drive_count / 2 {
|
||||
let error = store
|
||||
.get_bucket_info_from_sets(&bucket, &BucketOptions::default())
|
||||
.await
|
||||
.expect_err("bucket mutations must retain their majority namespace check");
|
||||
assert_eq!(error, StorageError::ErasureWriteQuorum);
|
||||
}
|
||||
|
||||
let below_quorum = take_set_disks_offline(&store, set, &[parity]).await;
|
||||
let read = store
|
||||
.get_object_reader(&bucket, object, None, Default::default(), &ObjectOptions::default())
|
||||
.await;
|
||||
match read {
|
||||
Ok(mut reader) => assert!(
|
||||
reader.stream.read_to_end(&mut Vec::new()).await.is_err(),
|
||||
"layout {drive_count}/{parity} must reject fewer than its data-shard quorum"
|
||||
),
|
||||
Err(error) => assert!(
|
||||
matches!(error, StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _)),
|
||||
"a missing shard must report read quorum loss, got {error}"
|
||||
),
|
||||
}
|
||||
restore_set_disks(&store, set, below_quorum).await;
|
||||
restore_set_disks(&store, set, offline).await;
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn bucket_info_read_quorum_is_scoped_to_each_erasure_set() {
|
||||
let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4, 6], None).await;
|
||||
metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
|
||||
let bucket = "read-quorum-mixed-pools";
|
||||
store
|
||||
.make_bucket(bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("healthy pools should accept bucket creation");
|
||||
|
||||
let first_set = &store.pools[0].disk_set[0];
|
||||
let second_set = &store.pools[1].disk_set[0];
|
||||
let first_offline = take_set_disks_offline(&store, first_set, &[0, 1]).await;
|
||||
let second_offline = take_set_disks_offline(&store, second_set, &[0, 1, 2]).await;
|
||||
store
|
||||
.get_bucket_info(bucket, &BucketOptions::default())
|
||||
.await
|
||||
.expect("each set independently satisfies its namespace read quorum");
|
||||
|
||||
for (set, extra_disk) in [(first_set, 2), (second_set, 3)] {
|
||||
let extra_offline = take_set_disks_offline(&store, set, &[extra_disk]).await;
|
||||
assert_eq!(
|
||||
store
|
||||
.get_bucket_info(bucket, &BucketOptions::default())
|
||||
.await
|
||||
.expect_err("another pool must not subsidize a set below its read quorum"),
|
||||
StorageError::ErasureReadQuorum
|
||||
);
|
||||
restore_set_disks(&store, set, extra_offline).await;
|
||||
}
|
||||
restore_set_disks(&store, first_set, first_offline).await;
|
||||
restore_set_disks(&store, second_set, second_offline).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn bucket_info_read_quorum_requires_authoritative_metadata() {
|
||||
for state in ["missing", "corrupt", "foreign", "incarnation"] {
|
||||
let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4], None).await;
|
||||
metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
|
||||
let bucket = format!("read-quorum-{state}-metadata");
|
||||
let mut metadata = if state == "missing" {
|
||||
store
|
||||
.make_bucket_on_sets(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("simulate directories left before bucket metadata is published");
|
||||
BucketMetadata::new(&bucket)
|
||||
} else {
|
||||
store
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("healthy bucket should publish metadata");
|
||||
metadata_sys::get_in(&store.ctx, &bucket)
|
||||
.await
|
||||
.expect("seed metadata should be cached")
|
||||
.as_ref()
|
||||
.clone()
|
||||
};
|
||||
let path = metadata.save_file_path();
|
||||
match state {
|
||||
"corrupt" => crate::config::com::save_config(store.clone(), &path, b"corrupt".to_vec())
|
||||
.await
|
||||
.expect("persist corrupt metadata while the cached copy remains valid"),
|
||||
"foreign" => {
|
||||
metadata.name = "different-bucket".to_string();
|
||||
let mut encoded = vec![1, 0, 1, 0];
|
||||
encoded.extend(metadata.marshal_msg().expect("foreign metadata should encode"));
|
||||
crate::config::com::save_config(store.clone(), &path, encoded)
|
||||
.await
|
||||
.expect("persist metadata for a different bucket at the requested path");
|
||||
}
|
||||
"incarnation" => crate::bucket::metadata::save_bucket_incarnation(store.clone(), &bucket, Uuid::new_v4())
|
||||
.await
|
||||
.expect("persist a different bucket generation"),
|
||||
_ => {}
|
||||
}
|
||||
|
||||
let set = &store.pools[0].disk_set[0];
|
||||
let offline = take_set_disks_offline(&store, set, &[0, 1]).await;
|
||||
let error = store
|
||||
.get_bucket_info(&bucket, &BucketOptions::default())
|
||||
.await
|
||||
.expect_err("read admission must not trust residual directories or cached metadata");
|
||||
match state {
|
||||
"missing" => assert_eq!(error, StorageError::ErasureReadQuorum),
|
||||
"foreign" => assert_eq!(error, StorageError::FileCorrupt),
|
||||
"incarnation" => assert!(error.to_string().contains("sidecar does not match bucket metadata")),
|
||||
"corrupt" => assert!(error.to_string().contains("format invalid"), "unexpected corruption error: {error}"),
|
||||
_ => unreachable!(),
|
||||
}
|
||||
restore_set_disks(&store, set, offline).await;
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn bucket_info_read_quorum_accepts_persisted_legacy_metadata() {
|
||||
let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4], None).await;
|
||||
metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
|
||||
let bucket = "interop";
|
||||
store
|
||||
.make_bucket_on_sets(bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("legacy bucket directories should exist");
|
||||
let hex = include_str!("../../tests/fixtures/minio/bucket_metadata.blob.hex")
|
||||
.split_whitespace()
|
||||
.collect::<String>();
|
||||
let body = (0..hex.len())
|
||||
.step_by(2)
|
||||
.map(|index| u8::from_str_radix(&hex[index..index + 2], 16).expect("pinned MinIO metadata fixture"))
|
||||
.collect();
|
||||
crate::config::com::save_config(store.clone(), &BucketMetadata::new(bucket).save_file_path(), body)
|
||||
.await
|
||||
.expect("legacy metadata should be persisted without an incarnation sidecar");
|
||||
|
||||
let set = &store.pools[0].disk_set[0];
|
||||
let offline = take_set_disks_offline(&store, set, &[0, 1]).await;
|
||||
let info = store
|
||||
.get_bucket_info(bucket, &BucketOptions::default())
|
||||
.await
|
||||
.expect("persisted MinIO metadata should authorize reads at the namespace read quorum");
|
||||
assert_eq!(info.name, bucket);
|
||||
assert!(info.versioning);
|
||||
assert!(info.object_locking);
|
||||
restore_set_disks(&store, set, offline).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn bucket_namespace_reads_report_missing_when_every_set_is_absent() {
|
||||
@@ -2207,6 +2476,7 @@ mod tests {
|
||||
#[serial]
|
||||
async fn bucket_namespace_reads_fail_closed_when_any_set_loses_quorum() {
|
||||
let (_temp_dir, ecstore) = setup_multi_pool_bucket_test_env().await;
|
||||
metadata_sys::init_bucket_metadata_sys(ecstore.clone(), Vec::new()).await;
|
||||
let bucket = format!("degraded-expansion-{}", Uuid::new_v4().simple());
|
||||
ecstore.pools[0].disk_set[0]
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
@@ -2214,6 +2484,7 @@ mod tests {
|
||||
.expect("bucket should be created in the original pool only");
|
||||
ecstore.pools[1].disk_set[0].disks.write().await[0] = None;
|
||||
ecstore.pools[1].disk_set[0].disks.write().await[1] = None;
|
||||
ecstore.pools[1].disk_set[0].disks.write().await[2] = None;
|
||||
|
||||
let list_err = ecstore
|
||||
.list_bucket(&BucketOptions::default())
|
||||
@@ -2225,7 +2496,7 @@ mod tests {
|
||||
.get_bucket_info(&bucket, &BucketOptions::default())
|
||||
.await
|
||||
.expect_err("bucket validation must fail when an expansion pool is unavailable");
|
||||
assert_eq!(info_err, StorageError::ErasureWriteQuorum);
|
||||
assert_eq!(info_err, StorageError::ErasureReadQuorum);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -577,13 +577,16 @@ impl KmsServiceManager {
|
||||
Some(service_version.probe_worker.as_ref()?.status())
|
||||
}
|
||||
|
||||
/// Health check for the KMS service
|
||||
/// Check backend health without changing the service lifecycle state.
|
||||
///
|
||||
/// A transient backend failure leaves the published service available for
|
||||
/// subsequent checks and operations. Readiness uses the background probe
|
||||
/// to evaluate backend availability independently of lifecycle state.
|
||||
pub async fn health_check(&self) -> Result<bool> {
|
||||
let checked_state = self.state.load_full();
|
||||
match checked_state.current_service.as_ref() {
|
||||
Some(service_version) => {
|
||||
let manager = service_version.manager.clone();
|
||||
let checked_version = service_version.version;
|
||||
// Perform health check on the backend
|
||||
match manager.health_check().await {
|
||||
Ok(healthy) => {
|
||||
@@ -594,8 +597,6 @@ impl KmsServiceManager {
|
||||
}
|
||||
Err(e) => {
|
||||
error!("KMS health check error: {}", e);
|
||||
let _guard = self.lifecycle_mutex.lock().await;
|
||||
self.mark_health_error_if_current(checked_version, &e);
|
||||
Err(e)
|
||||
}
|
||||
}
|
||||
@@ -739,17 +740,6 @@ impl KmsServiceManager {
|
||||
task: std::sync::Mutex::new(Some(task)),
|
||||
}))
|
||||
}
|
||||
|
||||
fn mark_health_error_if_current(&self, checked_version: u64, error: &KmsError) {
|
||||
let current = self.state.load_full();
|
||||
if current.current_service.as_ref().map(|version| version.version) == Some(checked_version) {
|
||||
self.state.store(Arc::new(RuntimeState {
|
||||
config: current.config.clone(),
|
||||
status: KmsServiceStatus::Error(format!("Health check failed: {error}")),
|
||||
current_service: current.current_service.clone(),
|
||||
}));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for KmsServiceManager {
|
||||
@@ -1004,19 +994,6 @@ mod tests {
|
||||
assert!(manager.get_service_version().await.expect("restarted version") > first_version);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stale_health_failure_cannot_poison_new_service_status() {
|
||||
let manager = KmsServiceManager::new();
|
||||
manager.configure(static_config("key-a", 0x11)).await.expect("configure");
|
||||
manager.start().await.expect("start");
|
||||
let old_version = manager.get_service_version().await.expect("old version");
|
||||
manager.restart().await.expect("restart");
|
||||
|
||||
manager.mark_health_error_if_current(old_version, &KmsError::backend_error("stale failure"));
|
||||
|
||||
assert_eq!(manager.get_status().await, KmsServiceStatus::Running);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn forbidden_local_master_key_change_preserves_running_config_and_service() {
|
||||
use crate::types::{CreateKeyRequest, KeyUsage};
|
||||
|
||||
@@ -75,6 +75,44 @@ fn unreachable_vault_config() -> KmsConfig {
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn transient_health_failure_does_not_latch_the_service_status() {
|
||||
let kms = TestKms::local().await;
|
||||
let manager = kms.manager();
|
||||
let service = manager.get_encryption_service().await.expect("running service");
|
||||
let version = manager.get_service_version().await.expect("running version");
|
||||
assert!(manager.health_check().await.expect("initial backend health"));
|
||||
|
||||
// Move only this test's keys out of reach, then restore the same backend.
|
||||
let key_dir = kms.key_dir().expect("local key directory");
|
||||
let outage = tempfile::TempDir::new().expect("temporary outage directory");
|
||||
let hidden_keys = outage.path().join("keys");
|
||||
tokio::fs::rename(&key_dir, &hidden_keys)
|
||||
.await
|
||||
.expect("make backend unavailable");
|
||||
let failure = manager.health_check().await;
|
||||
let outage_status = manager.get_status().await;
|
||||
tokio::fs::rename(&hidden_keys, &key_dir).await.expect("restore backend");
|
||||
|
||||
assert!(failure.is_err(), "the outage must surface as a health-check error");
|
||||
assert!(manager.health_check().await.expect("backend recovers without restart"));
|
||||
assert!(Arc::ptr_eq(
|
||||
&service,
|
||||
&manager.get_encryption_service().await.expect("service survives the outage")
|
||||
));
|
||||
assert_eq!(manager.get_service_version().await, Some(version));
|
||||
assert_eq!(
|
||||
manager.get_status().await,
|
||||
KmsServiceStatus::Running,
|
||||
"a recovered backend must not leave service-status and readiness latched in Error"
|
||||
);
|
||||
assert_eq!(
|
||||
outage_status,
|
||||
KmsServiceStatus::Running,
|
||||
"backend health does not change the running service's lifecycle state"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn starting_against_an_unreachable_backend_fails_without_publishing_a_service() {
|
||||
let manager = KmsServiceManager::new();
|
||||
|
||||
Reference in New Issue
Block a user