mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-26 22:07:53 +00:00
test(e2e): harden multipart fault and TLS acceptance (#8907)
* test(e2e): tighten scoped multipart fault acceptance Target Abort root deletion instead of child transaction cleanup, require actual metadata peer hits, and keep faulted APIs single-attempt. Validate bounded ordinary recovery and safe quorum refusal for partial mutations, with the recovery enhancement tracked in backlog#2940. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(e2e): drive concurrent TLS reads independently Give each coordinator its own supervised task and cooperative polling budget while retaining the existing workload, deadline and byte/hash assertions. Preserve node logs after fixture teardown and report the operation, response phase and received bytes when a transfer times out. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Hauser
@@ -1,2 +1,2 @@
|
||||
sha256-darwin=f21d495ca0c780f1f6e6bc626065599e3f16ea152bdc3ec08403af2fdf8b1203
|
||||
sha256-linux=609fca1f70ab7c58e2a6c3839a6dcb5abe8177d478cd243ed679a12d20eaedfa
|
||||
sha256-darwin=b53fe6d46cdffedfad6a83da0dfffa9aec10cd7c8ace53469f129587bc9f4d50
|
||||
sha256-linux=6be90f5c3886e9852ac27d51f2d47e6d2fa8bb424e0d03ee931268dcd0020db8
|
||||
|
||||
@@ -4,12 +4,13 @@
|
||||
use super::harness::{DistCluster, DistLayout, TestResult, unique_bucket};
|
||||
use crate::tls_gen;
|
||||
use aws_sdk_s3::{Client, config::retry::RetryConfig, primitives::ByteStream};
|
||||
use futures::future::try_join_all;
|
||||
use rand::{Rng, SeedableRng, rngs::StdRng};
|
||||
use sha2::{Digest, Sha256};
|
||||
use std::io::{Read, Seek, SeekFrom, Write};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
use tokio::task::JoinSet;
|
||||
|
||||
const MIB: u64 = 1024 * 1024;
|
||||
const READ_TIMEOUT: Duration = Duration::from_secs(180);
|
||||
@@ -85,8 +86,10 @@ async fn verified_get(
|
||||
expected: &ExpectedRead,
|
||||
concurrent: bool,
|
||||
) -> TestResult<serde_json::Value> {
|
||||
let started = Instant::now();
|
||||
let mut received = 0u64;
|
||||
let mut stage = "response headers";
|
||||
tokio::time::timeout(READ_TIMEOUT, async {
|
||||
let started = Instant::now();
|
||||
let mut request = client.get_object().bucket(bucket).key(&object.key);
|
||||
if let Some((start, end)) = expected.range {
|
||||
request = request.range(format!("bytes={start}-{end}"));
|
||||
@@ -98,24 +101,37 @@ async fn verified_get(
|
||||
} else {
|
||||
assert_eq!(output.content_range(), None);
|
||||
}
|
||||
let mut count = 0u64;
|
||||
stage = "response body";
|
||||
let mut digest = Sha256::new();
|
||||
while let Some(chunk) = output.body.next().await {
|
||||
let chunk = chunk?;
|
||||
count += u64::try_from(chunk.len())?;
|
||||
assert!(count <= expected.length, "node {node}: GET exceeded the requested length");
|
||||
received += u64::try_from(chunk.len())?;
|
||||
assert!(received <= expected.length, "node {node}: GET exceeded the requested length");
|
||||
digest.update(&chunk);
|
||||
}
|
||||
let actual_sha256 = hex_simd::encode_to_string(digest.finalize(), hex_simd::AsciiCase::Lower);
|
||||
assert_eq!(count, expected.length, "node {node}: truncated GET {}", object.key);
|
||||
assert_eq!(received, expected.length, "node {node}: truncated GET {}", object.key);
|
||||
assert_eq!(actual_sha256, expected.sha256, "node {node}: corrupt GET {}", object.key);
|
||||
eprintln!(
|
||||
"TLS acceptance: GET node={node}, key={}, range={:?}, concurrent={concurrent}, bytes={received}, elapsed={:?}",
|
||||
object.key, expected.range, started.elapsed()
|
||||
);
|
||||
Ok(serde_json::json!({
|
||||
"operation": "get", "node": node, "key": object.key,
|
||||
"range": expected.range, "bytes": count, "sha256": actual_sha256,
|
||||
"range": expected.range, "bytes": received, "sha256": actual_sha256,
|
||||
"concurrent": concurrent, "elapsed_ms": started.elapsed().as_secs_f64() * 1000.0,
|
||||
}))
|
||||
})
|
||||
.await?
|
||||
.await
|
||||
.map_err(|error| {
|
||||
std::io::Error::new(
|
||||
std::io::ErrorKind::TimedOut,
|
||||
format!(
|
||||
"TLS GET timed out: node={node}, key={}, range={:?}, concurrent={concurrent}, stage={stage}, received={received}/{}, elapsed={:?}: {error}",
|
||||
object.key, expected.range, expected.length, started.elapsed()
|
||||
),
|
||||
)
|
||||
})?
|
||||
}
|
||||
|
||||
/// This case always exercises the full acceptance sizes; do not substitute
|
||||
@@ -157,9 +173,13 @@ async fn four_node_tls_large_and_range_gets_preserve_bytes() -> TestResult {
|
||||
{
|
||||
return Err("acceptance report paths must be unused".into());
|
||||
}
|
||||
let log_directory = report_path
|
||||
.as_ref()
|
||||
.map_or_else(|| directory.join("logs"), |path| path.with_extension("logs"));
|
||||
let log_directory = report_path.as_ref().map_or_else(
|
||||
|| {
|
||||
std::env::var_os("RUSTFS_E2E_LOG_DIR")
|
||||
.map_or_else(|| directory.join("logs"), |root| PathBuf::from(root).join(unique_bucket("internode-tls")))
|
||||
},
|
||||
|path| path.with_extension("logs"),
|
||||
);
|
||||
std::fs::create_dir(&log_directory)?;
|
||||
dist.capture_logs_under(&log_directory)?;
|
||||
dist.cluster.start().await?;
|
||||
@@ -183,7 +203,8 @@ async fn four_node_tls_large_and_range_gets_preserve_bytes() -> TestResult {
|
||||
let mut results = Vec::new();
|
||||
for length in [32 * 1024, 100 * MIB, 1024 * MIB] {
|
||||
let fixture_directory = directory.clone();
|
||||
let object = tokio::task::spawn_blocking(move || fixture(&fixture_directory, length)).await??;
|
||||
let object = Arc::new(tokio::task::spawn_blocking(move || fixture(&fixture_directory, length)).await??);
|
||||
eprintln!("TLS acceptance: fixture key={}, bytes={length}", object.key);
|
||||
let started = Instant::now();
|
||||
tokio::time::timeout(
|
||||
READ_TIMEOUT,
|
||||
@@ -194,7 +215,18 @@ async fn four_node_tls_large_and_range_gets_preserve_bytes() -> TestResult {
|
||||
.body(ByteStream::from_path(&object.path).await?)
|
||||
.send(),
|
||||
)
|
||||
.await??;
|
||||
.await
|
||||
.map_err(|error| {
|
||||
std::io::Error::new(
|
||||
std::io::ErrorKind::TimedOut,
|
||||
format!(
|
||||
"TLS PUT timed out: key={}, bytes={length}, elapsed={:?}: {error}",
|
||||
object.key,
|
||||
started.elapsed()
|
||||
),
|
||||
)
|
||||
})??;
|
||||
eprintln!("TLS acceptance: PUT key={}, bytes={length}, elapsed={:?}", object.key, started.elapsed());
|
||||
results.push(
|
||||
serde_json::json!({"operation": "put", "bytes": length, "elapsed_ms": started.elapsed().as_secs_f64() * 1000.0}),
|
||||
);
|
||||
@@ -216,15 +248,19 @@ async fn four_node_tls_large_and_range_gets_preserve_bytes() -> TestResult {
|
||||
results.push(verified_get(client, node, &bucket, &object, expected, false).await?);
|
||||
}
|
||||
}
|
||||
results.extend(
|
||||
try_join_all(
|
||||
clients
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(node, client)| verified_get(client, node, &bucket, &object, &object.reads[0], true)),
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
// Each reader owns a Tokio task and its cooperative budget. Polling
|
||||
// every body and synchronous digest in one parent task can let an
|
||||
// always-ready reader starve its siblings and their deadlines.
|
||||
let mut readers = JoinSet::new();
|
||||
for (node, client) in clients.iter().enumerate() {
|
||||
let client = client.clone();
|
||||
let bucket = bucket.clone();
|
||||
let object = Arc::clone(&object);
|
||||
readers.spawn(async move { verified_get(&client, node, &bucket, &object, &object.reads[0], true).await });
|
||||
}
|
||||
while let Some(result) = readers.join_next().await {
|
||||
results.push(result??);
|
||||
}
|
||||
tokio::fs::remove_file(&object.path).await?;
|
||||
}
|
||||
for node in 0..clients.len() {
|
||||
|
||||
@@ -13,7 +13,8 @@
|
||||
// limitations under the License.
|
||||
|
||||
use super::harness::{
|
||||
TestResult, assert_object_bytes, cluster_admin_ok, configure_node_logs, payload_for, retrying_put, unique_bucket, wait_until,
|
||||
TestResult, assert_object_bytes, cluster_admin_ok, configure_node_logs, payload_for, retrying_put, sha256_hex, unique_bucket,
|
||||
wait_until,
|
||||
};
|
||||
use crate::common::{ClusterTopology, RustFSTestClusterEnvironment};
|
||||
use crate::rpc_fault_proxy::{RpcFaultProxy, RpcPause};
|
||||
@@ -86,12 +87,19 @@ fn create(
|
||||
tokio::spawn(async move { request.send().await })
|
||||
}
|
||||
|
||||
async fn wait_for_faults(pauses: &[RpcPause]) -> TestResult {
|
||||
// Distributed locks use adaptive fan-out and may never contact an
|
||||
// unnecessary voter. Every selected peer remains gated; require a real
|
||||
// request hit without waiting for an unused peer to exhaust the S3 budget.
|
||||
let (result, _, _) = futures::future::select_all(pauses.iter().map(|pause| Box::pin(pause.wait_for_hits(1)))).await;
|
||||
result?;
|
||||
async fn wait_for_faults(pauses: &[RpcPause], methods: &[&str]) -> TestResult {
|
||||
if methods.iter().any(|method| matches!(*method, "Lock" | "LockBatch")) {
|
||||
// Adaptive lock fan-out may not contact an unnecessary voter. All
|
||||
// selected peers remain gated; require an actual scoped request hit.
|
||||
let (result, _, _) = futures::future::select_all(pauses.iter().map(|pause| Box::pin(pause.wait_for_hits(1)))).await;
|
||||
result?;
|
||||
} else {
|
||||
// Storage mutations fan out eagerly. Every selected peer must really
|
||||
// receive this upload's RPC, so an unrelated failure cannot fake quorum.
|
||||
for pause in pauses {
|
||||
pause.wait_for_hits(1).await?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -112,39 +120,14 @@ fn assert_unavailable<E: std::fmt::Debug + ProvideErrorMetadata>(error: &SdkErro
|
||||
eprintln!("MPU acceptance: cause={cause}, retry={retry:?}, request_id={request_id}");
|
||||
}
|
||||
|
||||
async fn abort(client: &Client, bucket: &str, key: &str, upload: &str, allow_missing: bool) -> TestResult {
|
||||
// Fault retirement resets outstanding RPC streams. Probe recovery within
|
||||
// the shared deadline while channels and late lock cleanup converge.
|
||||
wait_until(
|
||||
DEADLINE,
|
||||
|| async {
|
||||
match client
|
||||
.abort_multipart_upload()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.upload_id(upload)
|
||||
.send()
|
||||
.await
|
||||
{
|
||||
Ok(_) => Ok(true),
|
||||
Err(error) if allow_missing && error.as_service_error().and_then(|e| e.meta().code()) == Some("NoSuchUpload") => {
|
||||
Ok(true)
|
||||
}
|
||||
Err(error)
|
||||
if error.as_service_error().and_then(|e| e.meta().code()) == Some("ServiceUnavailable")
|
||||
&& error.raw_response().is_some_and(|response| {
|
||||
response.status().as_u16() == 503
|
||||
&& response.headers().get("x-rustfs-error-cause") == Some("namespace_lock_timeout")
|
||||
}) =>
|
||||
{
|
||||
Ok(false)
|
||||
}
|
||||
Err(error) => Err(error.into()),
|
||||
}
|
||||
},
|
||||
"multipart abort after RPC fault retirement",
|
||||
)
|
||||
.await?;
|
||||
async fn abort(client: &Client, bucket: &str, key: &str, upload: &str) -> TestResult {
|
||||
client
|
||||
.abort_multipart_upload()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.upload_id(upload)
|
||||
.send()
|
||||
.await?;
|
||||
let missing = client
|
||||
.list_parts()
|
||||
.bucket(bucket)
|
||||
@@ -157,6 +140,83 @@ async fn abort(client: &Client, bucket: &str, key: &str, upload: &str, allow_mis
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Fault removal does not synchronously clear existing server leases or drive
|
||||
// health state. Keep the injected operation single-attempt, then require
|
||||
// bounded recovery of the exact cleanup API instead of assuming instant heal.
|
||||
async fn recover_after_fault<T, E, F, Fut>(operation: &str, mut call: F) -> TestResult<Result<T, SdkError<E>>>
|
||||
where
|
||||
E: std::fmt::Debug + ProvideErrorMetadata,
|
||||
F: FnMut() -> Fut,
|
||||
Fut: std::future::Future<Output = Result<T, SdkError<E>>>,
|
||||
{
|
||||
let started = Instant::now();
|
||||
let mut attempts = 0;
|
||||
let mut last_failure = String::new();
|
||||
let outcome = tokio::time::timeout(DEADLINE, async {
|
||||
loop {
|
||||
attempts += 1;
|
||||
let result = call().await;
|
||||
let retry = result.as_ref().err().and_then(|error| {
|
||||
let response = error.raw_response()?;
|
||||
if response.status().as_u16() != 503
|
||||
|| error.as_service_error()?.meta().code() != Some("ServiceUnavailable")
|
||||
|| !matches!(
|
||||
response.headers().get("x-rustfs-error-cause"),
|
||||
Some("namespace_lock_timeout" | "namespace_lock_unavailable" | "storage_write_quorum")
|
||||
)
|
||||
{
|
||||
return None;
|
||||
}
|
||||
let delay = match response.headers().get("retry-after") {
|
||||
Some("1") => Duration::from_secs(1),
|
||||
None => Duration::from_millis(100),
|
||||
Some(_) => return None,
|
||||
};
|
||||
last_failure = format!(
|
||||
"cause={:?}, request_id={:?}",
|
||||
response.headers().get("x-rustfs-error-cause"),
|
||||
response.headers().get("x-amz-request-id")
|
||||
);
|
||||
Some(delay)
|
||||
});
|
||||
match retry {
|
||||
Some(delay) => tokio::time::sleep(delay).await,
|
||||
None => return result,
|
||||
}
|
||||
}
|
||||
})
|
||||
.await;
|
||||
eprintln!(
|
||||
"MPU recovery: operation={operation}, attempts={attempts}, elapsed={:?}, last={last_failure}",
|
||||
started.elapsed()
|
||||
);
|
||||
outcome.map_err(|_| format!("{operation} did not recover within {DEADLINE:?}: {last_failure}").into())
|
||||
}
|
||||
|
||||
async fn abort_after_fault(client: &Client, bucket: &str, key: &str, upload: &str, allow_missing: bool) -> TestResult {
|
||||
match recover_after_fault(&format!("Abort {key}"), || {
|
||||
client
|
||||
.abort_multipart_upload()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.upload_id(upload)
|
||||
.send()
|
||||
})
|
||||
.await?
|
||||
{
|
||||
Ok(_) => {}
|
||||
Err(error) if allow_missing && error.as_service_error().and_then(|e| e.meta().code()) == Some("NoSuchUpload") => {}
|
||||
Err(error) => return Err(error.into()),
|
||||
}
|
||||
let missing = recover_after_fault(&format!("ListParts {key}"), || {
|
||||
client.list_parts().bucket(bucket).key(key).upload_id(upload).send()
|
||||
})
|
||||
.await?
|
||||
.expect_err("a recovered Abort must remove the upload");
|
||||
assert_eq!(missing.as_service_error().and_then(|e| e.meta().code()), Some("NoSuchUpload"));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn round_trip(clients: &[Client], bucket: &str, key: &str, size: usize) -> TestResult {
|
||||
let upload = clients[0].create_multipart_upload().bucket(bucket).key(key).send().await?;
|
||||
let upload_id = upload.upload_id().ok_or("missing upload ID")?;
|
||||
@@ -207,6 +267,31 @@ async fn assert_no_uploads(clients: &[Client], bucket: &str) -> TestResult {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn assert_no_uploads_after_fault(clients: &[Client], bucket: &str) -> TestResult {
|
||||
for client in clients {
|
||||
let uploads = recover_after_fault("ListMultipartUploads after fault", || {
|
||||
client.list_multipart_uploads().bucket(bucket).send()
|
||||
})
|
||||
.await??;
|
||||
assert!(uploads.uploads().is_empty(), "unexpected residual uploads: {:?}", uploads.uploads());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn assert_object_bytes_after_fault(client: &Client, bucket: &str, key: &str, expected: &[u8]) -> TestResult {
|
||||
let output = recover_after_fault(&format!("GetObject {key} after fault"), || {
|
||||
client.get_object().bucket(bucket).key(key).send()
|
||||
})
|
||||
.await??;
|
||||
// A successful response must have a complete, correct body. Streaming
|
||||
// failures are not retried or converted into a successful comparison.
|
||||
let bytes = output.body.collect().await?.into_bytes();
|
||||
if bytes.as_ref() != expected {
|
||||
return Err(format!("object {key} bytes differ: expected {} bytes, got {}", expected.len(), bytes.len()).into());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// This fixture inspects RustFS RPCs, so use the server-generated upload's
|
||||
// canonical suffix and fail explicitly if its encoding changes.
|
||||
fn upload_scope(upload_id: &str) -> TestResult<String> {
|
||||
@@ -218,6 +303,14 @@ fn upload_scope(upload_id: &str) -> TestResult<String> {
|
||||
Ok(suffix.to_owned())
|
||||
}
|
||||
|
||||
fn upload_path_scope(bucket: &str, key: &str, upload_id: &str) -> TestResult<String> {
|
||||
Ok(format!(
|
||||
"{}/{}",
|
||||
sha256_hex(format!("{bucket}/{key}").as_bytes()),
|
||||
upload_scope(upload_id)?
|
||||
))
|
||||
}
|
||||
|
||||
async fn with_paused_peers<T: Send + 'static>(
|
||||
proxies: &[RpcFaultProxy],
|
||||
methods: &[&'static str],
|
||||
@@ -231,7 +324,7 @@ async fn with_paused_peers<T: Send + 'static>(
|
||||
.map(|(proxy, limit)| proxy.pause(methods, *limit, Some(scope)))
|
||||
.collect::<Result<_, _>>()?;
|
||||
let task = tokio::spawn(request);
|
||||
wait_for_faults(&pauses).await?;
|
||||
wait_for_faults(&pauses, methods).await?;
|
||||
let result = tokio::time::timeout(DEADLINE, task).await??;
|
||||
for pause in &pauses {
|
||||
pause.discard_pending();
|
||||
@@ -324,13 +417,23 @@ async fn phase_quorum_matrix(
|
||||
let request = clients[0].create_multipart_upload().bucket(bucket).key(&key);
|
||||
let result = with_paused_peers(proxies, methods, &limits, bucket, async move { request.send().await }).await?;
|
||||
if let Some(upload) = quorum_outcome(result, unavailable == unavailable_at_quorum, locks)? {
|
||||
abort(&clients[0], bucket, &key, upload.upload_id().ok_or("missing upload ID")?, false).await?;
|
||||
abort_after_fault(&clients[0], bucket, &key, upload.upload_id().ok_or("missing upload ID")?, false).await?;
|
||||
}
|
||||
} else {
|
||||
let upload = clients[0].create_multipart_upload().bucket(bucket).key(&key).send().await?;
|
||||
let upload = clients[0]
|
||||
.create_multipart_upload()
|
||||
.bucket(bucket)
|
||||
.key(&key)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|error| format!("setup Create failed for {key}: {error:?}"))?;
|
||||
let upload_id = upload.upload_id().ok_or("missing upload ID")?;
|
||||
let scope = upload_scope(upload_id)?;
|
||||
let body = payload_for(&key, 1024 * 1024);
|
||||
let scope = if !locks && phase == "abort" {
|
||||
upload_path_scope(bucket, &key, upload_id)?
|
||||
} else {
|
||||
upload_scope(upload_id)?
|
||||
};
|
||||
let body = payload_for(&key, if drives_per_node == 4 { 9 * 1024 * 1024 } else { 1024 * 1024 });
|
||||
let initial = if phase != "part" {
|
||||
Some(
|
||||
clients[0]
|
||||
@@ -358,7 +461,7 @@ async fn phase_quorum_matrix(
|
||||
let result =
|
||||
with_paused_peers(proxies, methods, &limits, &scope, async move { request.send().await }).await?;
|
||||
if let Some(part) = quorum_outcome(result, unavailable == unavailable_at_quorum, locks)? {
|
||||
clients[0]
|
||||
let completion = clients[0]
|
||||
.complete_multipart_upload()
|
||||
.bucket(bucket)
|
||||
.key(&key)
|
||||
@@ -372,11 +475,11 @@ async fn phase_quorum_matrix(
|
||||
.build(),
|
||||
)
|
||||
.build(),
|
||||
)
|
||||
.send()
|
||||
.await?;
|
||||
);
|
||||
recover_after_fault(&format!("Complete {key} after part fault"), || completion.clone().send())
|
||||
.await??;
|
||||
for reader in clients {
|
||||
assert_object_bytes(reader, bucket, &key, &body).await?;
|
||||
assert_object_bytes_after_fault(reader, bucket, &key, &body).await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -401,7 +504,7 @@ async fn phase_quorum_matrix(
|
||||
with_paused_peers(proxies, methods, &limits, &scope, async move { request.send().await }).await?;
|
||||
if quorum_outcome(result, unavailable == unavailable_at_quorum, locks)?.is_some() {
|
||||
for reader in clients {
|
||||
assert_object_bytes(reader, bucket, &key, &body).await?;
|
||||
assert_object_bytes_after_fault(reader, bucket, &key, &body).await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -417,14 +520,17 @@ async fn phase_quorum_matrix(
|
||||
}
|
||||
_ => unreachable!(),
|
||||
}
|
||||
// A failed completion can be ambiguous to an S3 client.
|
||||
// Cleanup also accepts an already completed/aborted session.
|
||||
abort(&clients[0], bucket, &key, upload_id, true).await?;
|
||||
clients[0].delete_object().bucket(bucket).key(&key).send().await?;
|
||||
// Cleanup must recover even after a below-quorum mutation;
|
||||
// only an already retired session may return NoSuchUpload.
|
||||
abort_after_fault(&clients[0], bucket, &key, upload_id, true).await?;
|
||||
recover_after_fault(&format!("DeleteObject {key} after {phase}"), || {
|
||||
clients[0].delete_object().bucket(bucket).key(&key).send()
|
||||
})
|
||||
.await??;
|
||||
}
|
||||
wait_for_lock_peer_recovery(proxies, &clients[0], bucket).await?;
|
||||
round_trip(clients, bucket, "recovered", 9 * 1024 * 1024).await?;
|
||||
assert_no_uploads(clients, bucket).await?;
|
||||
assert_no_uploads_after_fault(clients, bucket).await?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
@@ -463,7 +569,7 @@ async fn cancelled_metadata_wait_releases_fence_for_bucket_recreation() -> TestR
|
||||
.map(|proxy| proxy.pause(&["WriteMetadata"], usize::MAX, Some(&bucket)))
|
||||
.collect::<Result<_, _>>()?;
|
||||
let task = create(&clients[0], &bucket, "cancelled");
|
||||
wait_for_faults(&pauses).await?;
|
||||
wait_for_faults(&pauses, &["WriteMetadata"]).await?;
|
||||
task.abort();
|
||||
assert!(task.await.expect_err("cancelled S3 future").is_cancelled());
|
||||
drop(pauses);
|
||||
@@ -547,7 +653,7 @@ async fn eight_node_eight_drive_multipart_load_leaves_no_residual_uploads() -> T
|
||||
.body(ByteStream::from(vec![0xa5; 8 * 1024 * 1024]))
|
||||
.send()
|
||||
.await?;
|
||||
abort(&clients[(round + 2) % clients.len()], &bucket, "aborted", upload_id, false).await?;
|
||||
abort(&clients[(round + 2) % clients.len()], &bucket, "aborted", upload_id).await?;
|
||||
assert_no_uploads(&clients, &bucket).await?;
|
||||
eprintln!(
|
||||
"MPU 8x8 acceptance: round={}/{rounds}, completed={}, aborted={}, elapsed={:?}",
|
||||
@@ -572,7 +678,7 @@ async fn resume_at_rpc<T: Send + 'static>(
|
||||
.collect::<Result<_, _>>()?;
|
||||
eprintln!("MPU delay acceptance: methods={methods:?}, scope={scope}");
|
||||
let task = tokio::spawn(request);
|
||||
wait_for_faults(&pauses).await?;
|
||||
wait_for_faults(&pauses, methods).await?;
|
||||
assert!(
|
||||
!task.is_finished(),
|
||||
"the selected commit must wait while only one shard/voter is available"
|
||||
@@ -649,6 +755,11 @@ async fn each_multipart_phase_resumes_after_metadata_and_lock_delays() -> TestRe
|
||||
let upload = client.create_multipart_upload().bucket(&bucket).key(&key).send().await?;
|
||||
let upload_id = upload.upload_id().ok_or("missing upload ID")?;
|
||||
let scope = upload_scope(upload_id)?;
|
||||
let scope = if locks {
|
||||
scope
|
||||
} else {
|
||||
upload_path_scope(&bucket, &key, upload_id)?
|
||||
};
|
||||
let request = client.abort_multipart_upload().bucket(&bucket).key(&key).upload_id(upload_id);
|
||||
resume_at_rpc(&proxies, if locks { &["Lock", "LockBatch"] } else { &["Delete"] }, &scope, async move {
|
||||
Ok(request.send().await?)
|
||||
|
||||
@@ -22,6 +22,8 @@ use hyper_util::{
|
||||
rt::{TokioExecutor, TokioIo},
|
||||
server::conn::auto,
|
||||
};
|
||||
use prost::Message as _;
|
||||
use rustfs_protos::proto_gen::node_service::DeleteRequest;
|
||||
use std::{
|
||||
io,
|
||||
net::SocketAddr,
|
||||
@@ -195,6 +197,30 @@ impl RpcFaultProxy {
|
||||
}
|
||||
}
|
||||
|
||||
fn scope_matches(method: &str, body: &[u8], scope: &[u8]) -> io::Result<bool> {
|
||||
if method != "Delete" {
|
||||
return Ok(body.windows(scope.len()).any(|window| window == scope));
|
||||
}
|
||||
// Abort's recursive root delete must not be confused with transaction or
|
||||
// part cleanup below that root. This is a unary, uncompressed gRPC frame.
|
||||
if body.len() < 5 || body[0] != 0 {
|
||||
return Err(io::Error::new(
|
||||
io::ErrorKind::InvalidData,
|
||||
"scoped Delete needs an uncompressed unary frame",
|
||||
));
|
||||
}
|
||||
let mut encoded_length = [0; 4];
|
||||
encoded_length.copy_from_slice(&body[1..5]);
|
||||
if u32::try_from(body.len() - 5).ok() != Some(u32::from_be_bytes(encoded_length)) {
|
||||
return Err(io::Error::new(io::ErrorKind::InvalidData, "scoped Delete frame length mismatch"));
|
||||
}
|
||||
let request = DeleteRequest::decode(&body[5..]).map_err(io::Error::other)?;
|
||||
let options: serde_json::Value = serde_json::from_str(&request.options).map_err(io::Error::other)?;
|
||||
Ok(request.volume == ".rustfs.sys/multipart"
|
||||
&& request.path.trim_end_matches('/').as_bytes() == scope
|
||||
&& options.get("recursive").and_then(serde_json::Value::as_bool) == Some(true))
|
||||
}
|
||||
|
||||
async fn forward_request(
|
||||
request: Request<Incoming>,
|
||||
client: ProxyClient,
|
||||
@@ -206,14 +232,14 @@ async fn forward_request(
|
||||
let mut observed = None;
|
||||
if let Some(pause) = pause {
|
||||
let method = parts.uri.path().strip_prefix("/node_service.NodeService/");
|
||||
if method.is_some_and(|name| pause.methods.contains(&name)) {
|
||||
if let Some(method) = method.filter(|name| pause.methods.contains(name)) {
|
||||
let matches = if let Some(scope) = pause.scope.as_deref() {
|
||||
// Scope the control packet to this bucket or upload without
|
||||
// decoding or changing bytes covered by the RPC signature.
|
||||
// Inspect bounded routing data while retaining the exact
|
||||
// original bytes covered by the RPC signature.
|
||||
let collected = Limited::new(body, 1024 * 1024).collect().await.map_err(io::Error::other)?;
|
||||
let trailers = collected.trailers().cloned();
|
||||
let bytes = collected.to_bytes();
|
||||
let matches = bytes.windows(scope.len()).any(|window| window == scope);
|
||||
let matches = scope_matches(method, &bytes, scope)?;
|
||||
body = Full::new(bytes)
|
||||
.map_err(|never| match never {})
|
||||
.with_trailers(async move { trailers.map(Ok::<_, io::Error>) })
|
||||
@@ -504,4 +530,43 @@ mod tests {
|
||||
.await
|
||||
.expect("scoped discard deadline");
|
||||
}
|
||||
#[test]
|
||||
fn delete_scope_matches_only_the_upload_root() {
|
||||
let frame = |volume: &str, path: &str| {
|
||||
let message = DeleteRequest {
|
||||
volume: volume.to_owned(),
|
||||
path: path.to_owned(),
|
||||
options: r#"{"recursive":true}"#.to_owned(),
|
||||
..Default::default()
|
||||
};
|
||||
let payload = message.encode_to_vec();
|
||||
let length = u32::try_from(payload.len()).expect("small control packet");
|
||||
let mut bytes = vec![0];
|
||||
bytes.extend_from_slice(&length.to_be_bytes());
|
||||
bytes.extend_from_slice(&payload);
|
||||
bytes
|
||||
};
|
||||
let root = b"hash/upload";
|
||||
assert!(scope_matches("Delete", &frame(".rustfs.sys/multipart", "hash/upload"), root).expect("root frame"));
|
||||
assert!(scope_matches("Delete", &frame(".rustfs.sys/multipart", "hash/upload/"), root).expect("directory frame"));
|
||||
for path in ["hash/upload/part.1", "hash/upload-other"] {
|
||||
assert!(!scope_matches("Delete", &frame(".rustfs.sys/multipart", path), root).expect("other path"));
|
||||
}
|
||||
assert!(!scope_matches("Delete", &frame(".rustfs.sys/tmp", "hash/upload"), root).expect("other volume"));
|
||||
let non_recursive = DeleteRequest {
|
||||
volume: ".rustfs.sys/multipart".to_owned(),
|
||||
path: "hash/upload".to_owned(),
|
||||
options: r#"{"recursive":false}"#.to_owned(),
|
||||
..Default::default()
|
||||
}
|
||||
.encode_to_vec();
|
||||
let mut non_recursive_frame = vec![0];
|
||||
non_recursive_frame.extend_from_slice(&u32::try_from(non_recursive.len()).expect("small frame").to_be_bytes());
|
||||
non_recursive_frame.extend_from_slice(&non_recursive);
|
||||
assert!(!scope_matches("Delete", &non_recursive_frame, root).expect("non-recursive frame"));
|
||||
assert!(
|
||||
scope_matches("Delete", &[0, 0, 0, 0, 1], root).is_err(),
|
||||
"truncation must not become a match"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -22,7 +22,7 @@ HTTP/1.1 uses more concurrent connections than multiplexed HTTP/2. Size file-des
|
||||
|
||||
For validation, compare complete 100 MiB and 1 GiB GETs through every coordinator, verify lengths and hashes, and include concurrent GET, scanner, and heal activity. Record latency, stall errors, CPU, connection counts, and file descriptors. Include small objects and uploads when assessing a deployment's performance. Local transport tests exercise TLS ALPN, shared HTTP/2 backpressure, full-body delivery, and HTTP/1.1 idle reuse; they do not replace a distributed cluster run.
|
||||
|
||||
The distributed E2E case below starts four processes with four drives each and a generated CA. Both S3 and volume endpoints use verified TLS. It streams 32 KiB, 100 MiB and 1 GiB uploads from files, checks complete and unaligned Range responses through every coordinator, and repeats full reads with all four coordinators active concurrently. Each GET checks Content-Length, Content-Range where applicable, the received byte count and SHA-256. SDK GET retries are disabled, and captured node logs must contain no internode body stalls.
|
||||
The distributed E2E case below starts four processes with four drives each and a generated CA. Both S3 and volume endpoints use verified TLS. It streams 32 KiB, 100 MiB and 1 GiB uploads from files, checks complete and unaligned Range responses through every coordinator, and repeats full reads with all four coordinators active concurrently. Each concurrent GET has its own supervised Tokio task and cooperative budget, so body hashing in one reader cannot consume the other readers' polling budget. Each GET retains its 180-second deadline and checks Content-Length, Content-Range where applicable, the received byte count and SHA-256. SDK GET retries are disabled, and captured node logs must contain no internode body stalls. Timeout errors identify PUT or GET, object size, coordinator/range where applicable, response stage and received bytes; completed operations are recorded as they finish. In CI, node logs remain beneath `RUSTFS_E2E_LOG_DIR` after fixture teardown.
|
||||
|
||||
```bash
|
||||
export RUSTUP_TOOLCHAIN=stable
|
||||
|
||||
@@ -112,7 +112,7 @@ The `distributed::multipart_fence_acceptance_test` cases separate remote storage
|
||||
|
||||
Lock-layer regressions also cover every four-node read-quorum ordering, an eight-node write quorum behind silent voters, and the initial fan-out bound for prompt replies. Distributed acquisition probes unstarted voters within its existing attempt budget so silent early peers cannot hide a reachable quorum. Vote thresholds, owner identities, heartbeat checks, and late-grant cleanup retain their existing contracts.
|
||||
|
||||
Faults match the test bucket or upload suffix in bounded control-packet bodies, preserving their signed bytes and trailers. Readiness requires an actual matching RPC, without waiting for voters that adaptive fan-out does not contact. Recovery cases release waiting packets while the request is active; quorum cases discard expired packets before restoring the peer. Replaying a packet after its caller has timed out is a separate late-delivery fault model.
|
||||
Faults match the test bucket or upload suffix in bounded control-packet bodies, preserving their signed bytes and trailers. Abort injection decodes only the unary Delete routing fields and matches the exact multipart root, excluding child transaction cleanup. Readiness requires an actual matching RPC, without waiting for voters that adaptive fan-out does not contact. Recovery cases release waiting packets while the request is active; quorum cases discard expired packets before restoring the peer. Replaying a packet after its caller has timed out is a separate late-delivery fault model. The injected S3 operation has SDK retries disabled. After the fault is removed, ordinary cleanup must recover within 45 seconds, allowing the existing lease and drive-health recovery intervals; only classified ServiceUnavailable responses are polled, and the server's Retry-After advice is honored. The fault-free eight-node workload keeps single-attempt APIs throughout. Failed Complete uses the best-effort rollback contract in [erasure-coding.md](../architecture/erasure-coding.md#7-write-path-and-write-quorum). A below-quorum Abort can also leave partially deleted staging. Each matrix reuses one cluster and bucket, requiring fresh responses from every lock peer before the next row so circuit-breaker history cannot reduce its available votes. After a partial mutation, ordinary Abort must succeed or report an already retired session, ListParts must return NoSuchUpload, and every node must list no residual uploads. This verifies the Abort recovery merged in #8904 without weakening the delete quorum; it does not claim residue-free rollback of a failed Complete. Bucket-generation recreation is covered by the cancellation case.
|
||||
|
||||
The four-node EC:2 cases cover exactly quorum and quorum-minus-one for storage writes and exclusive locks (three votes) and ordinary Create shared locks (two votes), error-cause and retry headers, each multipart phase, and bucket recreation after client cancellation. A sixteen-shard EC12+4 case repeats all four metadata commit phases at exactly twelve and eleven available shards. The eight-node, eight-drive case checks the server-reported one-pool/four-set EC12+4 geometry, uploads 45 MiB objects using 8 MiB parts with two files and four parts in flight, verifies complete bytes through every node, and checks aborted and completed uploads leave no visible sessions.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user