mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-06 20:19:14 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| e32c7c4b65 | |||
| cf1c45eb91 |
@@ -3837,6 +3837,164 @@ async fn test_bucket_replication_converges_delete_marker_and_version_purge() ->
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Regression for rustfs/backlog#2340 (not Wasabi specific): a directory
|
||||
/// marker (`prefix/` with a body) in a versioned bucket is stored as the null
|
||||
/// version, like MinIO (`putOpts`: "for directory objects skip creating new
|
||||
/// versions"), and must still replicate to completion instead of staying
|
||||
/// `PENDING`.
|
||||
#[tokio::test]
|
||||
async fn test_bucket_replication_replicates_directory_marker_in_versioned_bucket() -> TestResult {
|
||||
init_logging();
|
||||
|
||||
let mut source_env = RustFSTestEnvironment::new().await?;
|
||||
let mut source_env_vars = replication_fast_env();
|
||||
source_env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
|
||||
source_env.start_rustfs_server_with_env(vec![], &source_env_vars).await?;
|
||||
|
||||
let mut target_env = RustFSTestEnvironment::new().await?;
|
||||
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
|
||||
|
||||
let source_bucket = "replication-dir-marker-src";
|
||||
let target_bucket = "replication-dir-marker-dst";
|
||||
let source_client = source_env.create_s3_client();
|
||||
let target_client = target_env.create_s3_client();
|
||||
|
||||
source_client.create_bucket().bucket(source_bucket).send().await?;
|
||||
target_client.create_bucket().bucket(target_bucket).send().await?;
|
||||
enable_bucket_versioning(&source_env, source_bucket).await?;
|
||||
enable_bucket_versioning(&target_env, target_bucket).await?;
|
||||
let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
|
||||
put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
|
||||
|
||||
let marker_key = "dir/trailing/";
|
||||
let body = b"directory marker body";
|
||||
let put = source_client
|
||||
.put_object()
|
||||
.bucket(source_bucket)
|
||||
.key(marker_key)
|
||||
.body(ByteStream::from_static(body))
|
||||
.send()
|
||||
.await?;
|
||||
assert!(
|
||||
put.version_id()
|
||||
.is_none_or(|id| id == "null" || id == uuid::Uuid::nil().to_string()),
|
||||
"a directory marker is the null version even in a versioned bucket: {:?}",
|
||||
put.version_id()
|
||||
);
|
||||
|
||||
wait_for_source_replication_status(&source_client, source_bucket, marker_key, "COMPLETED", false).await?;
|
||||
|
||||
let replica = target_client
|
||||
.get_object()
|
||||
.bucket(target_bucket)
|
||||
.key(marker_key)
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), body);
|
||||
let listed = target_client
|
||||
.list_object_versions()
|
||||
.bucket(target_bucket)
|
||||
.prefix(marker_key)
|
||||
.send()
|
||||
.await?;
|
||||
let marker_versions: Vec<_> = listed.versions().iter().filter(|v| v.key() == Some(marker_key)).collect();
|
||||
assert_eq!(marker_versions.len(), 1, "the marker must land exactly once: {marker_versions:?}");
|
||||
assert_eq!(
|
||||
marker_versions[0].version_id(),
|
||||
Some("null"),
|
||||
"the replica keeps the null version identity"
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Regression for rustfs/backlog#2340 (not Wasabi specific): a single-part
|
||||
/// object uploaded with `x-amz-checksum-*` must reach the target with the same
|
||||
/// checksum. The outbound options keyed the stored record by algorithm name,
|
||||
/// which the target client sent as `x-amz-meta-*` user metadata, so a replica
|
||||
/// never carried a checksum although the source HEAD returned one.
|
||||
#[tokio::test]
|
||||
async fn test_bucket_replication_forwards_single_part_object_checksums() -> TestResult {
|
||||
init_logging();
|
||||
|
||||
let mut source_env = RustFSTestEnvironment::new().await?;
|
||||
let mut source_env_vars = replication_fast_env();
|
||||
source_env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
|
||||
source_env.start_rustfs_server_with_env(vec![], &source_env_vars).await?;
|
||||
|
||||
let mut target_env = RustFSTestEnvironment::new().await?;
|
||||
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
|
||||
|
||||
let source_bucket = "replication-checksum-src";
|
||||
let target_bucket = "replication-checksum-dst";
|
||||
let source_client = source_env.create_s3_client();
|
||||
let target_client = target_env.create_s3_client();
|
||||
|
||||
source_client.create_bucket().bucket(source_bucket).send().await?;
|
||||
target_client.create_bucket().bucket(target_bucket).send().await?;
|
||||
enable_bucket_versioning(&source_env, source_bucket).await?;
|
||||
enable_bucket_versioning(&target_env, target_bucket).await?;
|
||||
let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
|
||||
put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
|
||||
|
||||
let body = b"123456789";
|
||||
let crc32_key = "checksum-crc32.txt";
|
||||
let sha256_key = "checksum-sha256.txt";
|
||||
let crc32_put = source_client
|
||||
.put_object()
|
||||
.bucket(source_bucket)
|
||||
.key(crc32_key)
|
||||
.body(ByteStream::from_static(body))
|
||||
.checksum_algorithm(aws_sdk_s3::types::ChecksumAlgorithm::Crc32)
|
||||
.send()
|
||||
.await?;
|
||||
let expected_crc32 = crc32_put.checksum_crc32().ok_or("source PUT omitted CRC32")?.to_string();
|
||||
let sha256_put = source_client
|
||||
.put_object()
|
||||
.bucket(source_bucket)
|
||||
.key(sha256_key)
|
||||
.body(ByteStream::from_static(body))
|
||||
.checksum_algorithm(aws_sdk_s3::types::ChecksumAlgorithm::Sha256)
|
||||
.send()
|
||||
.await?;
|
||||
let expected_sha256 = sha256_put.checksum_sha256().ok_or("source PUT omitted SHA256")?.to_string();
|
||||
|
||||
for key in [crc32_key, sha256_key] {
|
||||
wait_for_source_replication_status(&source_client, source_bucket, key, "COMPLETED", false).await?;
|
||||
}
|
||||
|
||||
let replica = target_client
|
||||
.head_object()
|
||||
.bucket(target_bucket)
|
||||
.key(crc32_key)
|
||||
.checksum_mode(aws_sdk_s3::types::ChecksumMode::Enabled)
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(replica.checksum_crc32(), Some(expected_crc32.as_str()), "replica lost the CRC32 checksum");
|
||||
let replica = target_client
|
||||
.head_object()
|
||||
.bucket(target_bucket)
|
||||
.key(sha256_key)
|
||||
.checksum_mode(aws_sdk_s3::types::ChecksumMode::Enabled)
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(
|
||||
replica.checksum_sha256(),
|
||||
Some(expected_sha256.as_str()),
|
||||
"replica lost the SHA256 checksum"
|
||||
);
|
||||
// The bare algorithm name must not leak as user metadata either.
|
||||
assert!(
|
||||
replica
|
||||
.metadata()
|
||||
.is_none_or(|meta| !meta.keys().any(|k| k.eq_ignore_ascii_case("sha256"))),
|
||||
"replica carries the checksum as user metadata: {:?}",
|
||||
replica.metadata()
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_bucket_replication_disabled_delete_marker_does_not_propagate() -> TestResult {
|
||||
init_logging();
|
||||
|
||||
@@ -42,7 +42,8 @@ use crate::replication_extension_test::{
|
||||
use aws_sdk_s3::Client;
|
||||
use aws_sdk_s3::primitives::{ByteStream, DateTime};
|
||||
use aws_sdk_s3::types::{
|
||||
Checksum, CompletedMultipartUpload, CompletedPart, ObjectAttributes, ObjectLockLegalHoldStatus, ObjectLockMode,
|
||||
Checksum, ChecksumAlgorithm, CompletedMultipartUpload, CompletedPart, ObjectAttributes, ObjectLockLegalHoldStatus,
|
||||
ObjectLockMode,
|
||||
};
|
||||
use bytes::Bytes;
|
||||
use std::error::Error;
|
||||
@@ -114,10 +115,13 @@ enum ObjectShape {
|
||||
LockedMultipart,
|
||||
/// ODM stores two local parts while preserving a single-PUT source's MD5 ETag.
|
||||
OdmPreservedMd5Multipart,
|
||||
/// Single-part object uploaded with `x-amz-checksum-sha256`; the replica
|
||||
/// must carry the same header (rustfs/backlog#2340).
|
||||
Checksummed,
|
||||
}
|
||||
|
||||
impl ObjectShape {
|
||||
const ALL: [ObjectShape; 7] = [
|
||||
const ALL: [ObjectShape; 8] = [
|
||||
ObjectShape::Empty,
|
||||
ObjectShape::Plain,
|
||||
ObjectShape::Retention,
|
||||
@@ -125,6 +129,7 @@ impl ObjectShape {
|
||||
ObjectShape::Multipart,
|
||||
ObjectShape::LockedMultipart,
|
||||
ObjectShape::OdmPreservedMd5Multipart,
|
||||
ObjectShape::Checksummed,
|
||||
];
|
||||
|
||||
fn key(self) -> &'static str {
|
||||
@@ -136,6 +141,16 @@ impl ObjectShape {
|
||||
ObjectShape::Multipart => "matrix/multipart.bin",
|
||||
ObjectShape::LockedMultipart => "matrix/locked-multipart.bin",
|
||||
ObjectShape::OdmPreservedMd5Multipart => "matrix/odm-preserved-md5.bin",
|
||||
ObjectShape::Checksummed => "matrix/checksummed.bin",
|
||||
}
|
||||
}
|
||||
|
||||
/// The `x-amz-checksum-*` header the source stored and every upload of
|
||||
/// the replica must repeat.
|
||||
fn forwarded_checksum_header(self) -> Option<&'static str> {
|
||||
match self {
|
||||
ObjectShape::Checksummed => Some("x-amz-checksum-sha256"),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -198,6 +213,18 @@ impl ObjectShape {
|
||||
ObjectShape::Multipart => multipart_put(client, bucket, key, 0x44, false).await,
|
||||
ObjectShape::LockedMultipart => multipart_put(client, bucket, key, 0x55, true).await,
|
||||
ObjectShape::OdmPreservedMd5Multipart => odm_preserved_md5_multipart(env, bucket, key).await,
|
||||
ObjectShape::Checksummed => {
|
||||
let body = payload(40 * 1024, 0x66);
|
||||
client
|
||||
.put_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.body(ByteStream::from(body.clone()))
|
||||
.checksum_algorithm(ChecksumAlgorithm::Sha256)
|
||||
.send()
|
||||
.await?;
|
||||
Ok(body)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -450,6 +477,19 @@ async fn check_completed_cell(
|
||||
}) {
|
||||
return Err(format!("a locked PutObject went out without any integrity header (rustfs#7082): {bare:?}").into());
|
||||
}
|
||||
// rustfs/backlog#2340 contract: a source checksum reaches the target as
|
||||
// the `x-amz-checksum-*` header, not as user metadata; every PutObject of
|
||||
// the shape carries it.
|
||||
if let Some(header) = shape.forwarded_checksum_header()
|
||||
&& let Some(missing) = uploads.iter().find(|record| {
|
||||
record.operation == FakeTargetOperation::PutObject
|
||||
&& !record.transport.checksum_headers.iter().any(|name| name == header)
|
||||
})
|
||||
{
|
||||
return Err(
|
||||
format!("a PutObject went out without the source's {header} header (rustfs/backlog#2340): {missing:?}").into(),
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -2064,7 +2064,15 @@ impl TargetClient {
|
||||
}
|
||||
}
|
||||
|
||||
match builder
|
||||
// A forwarded source checksum is this PUT's integrity header. In
|
||||
// streaming-checksum mode (`RUSTFS_REPLICATION_STREAMING_CHECKSUMS`)
|
||||
// the SDK would still add its default CRC32 trailer, and a target that
|
||||
// receives both keeps the trailer's algorithm: a forwarded SHA256
|
||||
// vanished from the replica while the source reported COMPLETED. Pin
|
||||
// this request to WhenRequired so nothing is sent beside the source's
|
||||
// own checksum.
|
||||
let forwards_source_checksum = headers.keys().any(|name| name.as_str().starts_with("x-amz-checksum-"));
|
||||
let mut operation = builder
|
||||
.bucket(bucket)
|
||||
.key(object)
|
||||
.content_length(size)
|
||||
@@ -2084,10 +2092,14 @@ impl TargetClient {
|
||||
}
|
||||
|
||||
Result::<_, aws_smithy_types::error::operation::BuildError>::Ok(req)
|
||||
})
|
||||
.send()
|
||||
.await
|
||||
{
|
||||
});
|
||||
if forwards_source_checksum {
|
||||
operation = operation.config_override(
|
||||
aws_sdk_s3::config::Builder::new()
|
||||
.request_checksum_calculation(aws_sdk_s3::config::RequestChecksumCalculation::WhenRequired),
|
||||
);
|
||||
}
|
||||
match operation.send().await {
|
||||
Ok(output) => {
|
||||
// Under SSE-KMS/DSSE or SSE-C the target's ETag is not the MD5
|
||||
// of the stored plaintext, so it cannot be compared against the
|
||||
@@ -2507,13 +2519,21 @@ mod tests {
|
||||
}
|
||||
|
||||
fn header_recording_target_client(response_headers: Vec<(String, String)>) -> (TargetClient, RecordedHeaders) {
|
||||
header_recording_target_client_with_checksums(response_headers, replication_request_checksum_calculation())
|
||||
}
|
||||
|
||||
fn header_recording_target_client_with_checksums(
|
||||
response_headers: Vec<(String, String)>,
|
||||
checksums: RequestChecksumCalculation,
|
||||
) -> (TargetClient, RecordedHeaders) {
|
||||
let request_headers: RecordedHeaders = Arc::new(std::sync::Mutex::new(Vec::new()));
|
||||
let connector = SharedHttpConnector::new(RecordingHeaderConnector {
|
||||
request_headers: Arc::clone(&request_headers),
|
||||
response_headers,
|
||||
});
|
||||
let http_client = http_client_fn(move |_settings, _components| connector.clone());
|
||||
let client = s3_client_for_test(443, Some(http_client));
|
||||
let client =
|
||||
s3_client_for_endpoint_test_with_checksums("https://localhost:443".to_string(), Some(http_client), checksums);
|
||||
(
|
||||
TargetClient {
|
||||
endpoint: "https://localhost:443".to_string(),
|
||||
@@ -2680,6 +2700,47 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
/// With streaming checksums enabled the SDK adds a CRC32 trailer to every
|
||||
/// upload. A PUT that forwards the source's checksum must not get that
|
||||
/// second algorithm: a target that receives both keeps the trailer's and
|
||||
/// the forwarded SHA256 never reaches the replica (rustfs/backlog#2340).
|
||||
#[tokio::test]
|
||||
async fn streaming_put_object_with_forwarded_checksum_sends_no_sdk_checksum() {
|
||||
let (client, recorded) =
|
||||
header_recording_target_client_with_checksums(Vec::new(), RequestChecksumCalculation::WhenSupported);
|
||||
let mut forwarded = PutObjectOptions::default();
|
||||
forwarded.user_metadata.insert(
|
||||
"x-amz-checksum-sha256".to_string(),
|
||||
"OoJ3yNhRwv3wwtZoGqEIPrPX9xwTnfLl+ka0wStN1g0=".to_string(),
|
||||
);
|
||||
client
|
||||
.put_object("target-bucket", "object", 4, streaming_test_body(b"data"), &forwarded)
|
||||
.await
|
||||
.expect("recorded put_object should succeed");
|
||||
client
|
||||
.put_object("target-bucket", "object", 4, streaming_test_body(b"data"), &PutObjectOptions::default())
|
||||
.await
|
||||
.expect("recorded put_object should succeed");
|
||||
let recorded = recorded.lock().expect("recorded header lock should not be poisoned");
|
||||
let with_forwarded = &recorded[0];
|
||||
assert_eq!(
|
||||
recorded_header(with_forwarded, "x-amz-checksum-sha256"),
|
||||
Some("OoJ3yNhRwv3wwtZoGqEIPrPX9xwTnfLl+ka0wStN1g0=")
|
||||
);
|
||||
assert_eq!(
|
||||
recorded_header(with_forwarded, "x-amz-trailer"),
|
||||
None,
|
||||
"the SDK must not add a trailer checksum"
|
||||
);
|
||||
assert_eq!(recorded_header(with_forwarded, "x-amz-sdk-checksum-algorithm"), None);
|
||||
// Control: the same client still streams a trailer when nothing is forwarded.
|
||||
let without_forwarded = &recorded[1];
|
||||
assert!(
|
||||
recorded_header(without_forwarded, "x-amz-trailer").is_some(),
|
||||
"streaming mode must still apply to uploads without a forwarded checksum: {without_forwarded:?}"
|
||||
);
|
||||
}
|
||||
|
||||
/// A forwarded source checksum already satisfies the rule; nothing is added.
|
||||
#[tokio::test]
|
||||
async fn locked_put_object_keeps_a_forwarded_source_checksum() {
|
||||
@@ -3045,6 +3106,14 @@ mod tests {
|
||||
}
|
||||
|
||||
fn s3_client_for_endpoint_test(endpoint: String, http_client: Option<SharedHttpClient>) -> S3Client {
|
||||
s3_client_for_endpoint_test_with_checksums(endpoint, http_client, replication_request_checksum_calculation())
|
||||
}
|
||||
|
||||
fn s3_client_for_endpoint_test_with_checksums(
|
||||
endpoint: String,
|
||||
http_client: Option<SharedHttpClient>,
|
||||
checksums: RequestChecksumCalculation,
|
||||
) -> S3Client {
|
||||
let credentials = SdkCredentials::builder()
|
||||
.access_key_id("test-access")
|
||||
.secret_access_key("test-secret")
|
||||
@@ -3058,7 +3127,7 @@ mod tests {
|
||||
.behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
|
||||
// Mirror the production remote-target builder so recorded requests
|
||||
// exercise the same checksum/framing behavior (#6853).
|
||||
.request_checksum_calculation(replication_request_checksum_calculation());
|
||||
.request_checksum_calculation(checksums);
|
||||
if let Some(http_client) = http_client {
|
||||
config = config.http_client(http_client);
|
||||
}
|
||||
|
||||
@@ -281,12 +281,6 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
|
||||
// record may only add multipart-ness, never take it away.
|
||||
is_multipart = base_is_multipart || checksum_record_is_multipart;
|
||||
|
||||
for (key, value) in checksum_meta.iter() {
|
||||
if key != AMZ_CHECKSUM_TYPE {
|
||||
meta.insert(key.clone(), value.clone());
|
||||
}
|
||||
}
|
||||
|
||||
if !base_is_multipart
|
||||
&& checksum_meta
|
||||
.get(AMZ_CHECKSUM_TYPE)
|
||||
@@ -294,6 +288,26 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
|
||||
{
|
||||
is_multipart = false;
|
||||
}
|
||||
|
||||
// The record keys each checksum by algorithm name ("CRC32"); the
|
||||
// target only reads `x-amz-checksum-<algorithm>`. Inserting the bare
|
||||
// name here made `PutObjectOptions::header()` send it as user
|
||||
// metadata (`x-amz-meta-crc32`), so no replica ever carried the
|
||||
// source checksum (rustfs/backlog#2340). The object-level record
|
||||
// describes one PUT body: a multipart replica is rebuilt part by
|
||||
// part, and its CreateMultipartUpload must not announce a checksum
|
||||
// the parts do not carry, so the record is forwarded on the
|
||||
// single-PUT route only (MinIO `getCRCMeta` parity).
|
||||
if !is_multipart {
|
||||
for (key, value) in checksum_meta.iter() {
|
||||
if key == AMZ_CHECKSUM_TYPE {
|
||||
continue;
|
||||
}
|
||||
if let Some(header) = rustfs_rio::ChecksumType::from_string(key).key() {
|
||||
meta.insert(header.to_string(), value.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1468,12 +1482,63 @@ mod tests {
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let (opts, _is_multipart) = replication_put_object_options("", &object_info).expect("build replication put options");
|
||||
let (opts, is_multipart) = replication_put_object_options("", &object_info).expect("build replication put options");
|
||||
|
||||
assert!(!is_multipart, "{name}: a single-part checksum record must keep the single-PUT route");
|
||||
let header = ty.key().expect("every forwarded algorithm has an x-amz-checksum header");
|
||||
assert_eq!(
|
||||
opts.user_metadata.get(name),
|
||||
opts.user_metadata.get(header),
|
||||
Some(&checksum.encoded),
|
||||
"replication must forward the {name} checksum into user_metadata identically to the classic algorithms"
|
||||
"replication must forward the {name} checksum as the {header} header"
|
||||
);
|
||||
assert!(
|
||||
!opts.user_metadata.contains_key(name),
|
||||
"{name}: the bare algorithm name would leave as x-amz-meta user metadata"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// The object-level record of a multipart upload (composite or full-object)
|
||||
/// must not become a PutObject checksum header: the replica is rebuilt
|
||||
/// through CreateMultipartUpload/UploadPart, and a checksum announced there
|
||||
/// that the parts do not carry would be rejected by the target.
|
||||
#[test]
|
||||
fn replication_put_object_options_keeps_multipart_checksum_records_off_the_wire() {
|
||||
let mut composite_type = rustfs_rio::ChecksumType::from_string("crc32");
|
||||
composite_type
|
||||
.merge(rustfs_rio::ChecksumType::MULTIPART)
|
||||
.merge(rustfs_rio::ChecksumType::INCLUDES_MULTIPART);
|
||||
let mut combined = Vec::new();
|
||||
for part in [b"part-one".as_slice(), b"part-two".as_slice()] {
|
||||
let part_checksum =
|
||||
rustfs_rio::Checksum::new_from_data(rustfs_rio::ChecksumType::from_string("crc32"), part).expect("part checksum");
|
||||
combined.extend_from_slice(part_checksum.raw.as_slice());
|
||||
}
|
||||
let composite = rustfs_rio::Checksum::new_from_data(composite_type, &combined)
|
||||
.expect("composite checksum")
|
||||
.to_bytes(&combined);
|
||||
|
||||
for (label, checksum, etag) in [
|
||||
("composite", composite, "0123456789abcdef0123456789abcdef-2"),
|
||||
(
|
||||
"full-object",
|
||||
full_object_multipart_checksum_record(),
|
||||
"0123456789abcdef0123456789abcdef-3",
|
||||
),
|
||||
] {
|
||||
let object_info = ObjectInfo {
|
||||
etag: Some(etag.to_string()),
|
||||
checksum: Some(checksum),
|
||||
..Default::default()
|
||||
};
|
||||
let (opts, is_multipart) = replication_put_object_options("", &object_info).expect("build replication put options");
|
||||
assert!(is_multipart, "{label}: a multipart object must keep the multipart route");
|
||||
assert!(
|
||||
opts.user_metadata
|
||||
.keys()
|
||||
.all(|key| !key.starts_with("x-amz-checksum-") && key != "CRC32"),
|
||||
"{label}: no object-level checksum may reach the target's CreateMultipartUpload: {:?}",
|
||||
opts.user_metadata
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -14,7 +14,6 @@
|
||||
|
||||
#[cfg(any(test, feature = "test-util"))]
|
||||
pub mod test_util;
|
||||
#[allow(clippy::module_inception, reason = "preserve the public services::tier::tier path")]
|
||||
pub mod tier;
|
||||
pub mod tier_admin;
|
||||
pub mod tier_config;
|
||||
|
||||
@@ -15,6 +15,8 @@
|
||||
#![allow(unused_variables)]
|
||||
#![allow(unused_mut)]
|
||||
#![allow(unused_assignments)]
|
||||
#![allow(unused_must_use)]
|
||||
#![allow(clippy::all)]
|
||||
|
||||
use byteorder::{ByteOrder, LittleEndian};
|
||||
use bytes::Bytes;
|
||||
@@ -800,7 +802,7 @@ pub enum TierConfigUpdateError {
|
||||
}
|
||||
|
||||
enum TierCandidateMutation {
|
||||
Add(Box<TierConfig>, bool),
|
||||
Add(TierConfig, bool),
|
||||
Edit(String, TierCreds),
|
||||
Remove(String, bool),
|
||||
Clear(bool),
|
||||
@@ -821,7 +823,7 @@ struct PrevalidatedTierCandidateMutation {
|
||||
impl TierCandidateMutation {
|
||||
fn add(mut config: TierConfig, force: bool) -> std::result::Result<Self, AdminError> {
|
||||
normalize_s3_gcs_add_tier_name(&mut config)?;
|
||||
Ok(Self::Add(Box::new(config), force))
|
||||
Ok(Self::Add(config, force))
|
||||
}
|
||||
|
||||
fn normalize_add_tier_name(&mut self) -> std::result::Result<(), AdminError> {
|
||||
@@ -908,7 +910,7 @@ impl TierCandidateMutation {
|
||||
match self {
|
||||
Self::Add(config, force) => {
|
||||
let tier_name = config.name.clone();
|
||||
candidate.add_with_deadline(*config, force, deadline).await?;
|
||||
candidate.add_with_deadline(config, force, deadline).await?;
|
||||
Ok(Some(tier_name))
|
||||
}
|
||||
Self::Edit(tier_name, credentials) => {
|
||||
@@ -2988,7 +2990,7 @@ fn from_external_tier_config(name: String, ext: ExternalTierConfig) -> io::Resul
|
||||
let tier_type = if wasabi_version {
|
||||
TierType::Wasabi
|
||||
} else {
|
||||
tier_type_from_hint(ext.tier_type_hint.as_deref()).unwrap_or(match ext.tier_type {
|
||||
tier_type_from_hint(ext.tier_type_hint.as_deref()).unwrap_or_else(|| match ext.tier_type {
|
||||
EXTERNAL_TIER_TYPE_S3 => TierType::S3,
|
||||
EXTERNAL_TIER_TYPE_AZURE => TierType::Azure,
|
||||
EXTERNAL_TIER_TYPE_GCS => TierType::GCS,
|
||||
@@ -3370,23 +3372,28 @@ impl TierConfigMgr {
|
||||
|
||||
pub async fn remove(&mut self, tier_name: &str, force: bool) -> std::result::Result<(), AdminError> {
|
||||
self.ensure_generation_is_idle(tier_name)?;
|
||||
let driver = match self.get_driver(tier_name).await {
|
||||
Ok(driver) => driver,
|
||||
Err(err) if err.code == ERR_TIER_NOT_FOUND.code => return Ok(()),
|
||||
Err(err) => return Err(err),
|
||||
};
|
||||
let d = self.get_driver(tier_name).await;
|
||||
if let Err(err) = d {
|
||||
if err.code == ERR_TIER_NOT_FOUND.code {
|
||||
return Ok(());
|
||||
} else {
|
||||
return Err(err);
|
||||
}
|
||||
}
|
||||
if !force {
|
||||
match driver.in_use().await {
|
||||
Err(err) => {
|
||||
let mut e = ERR_TIER_PERM_ERR.clone();
|
||||
e.message.push('.');
|
||||
e.message.push_str(&err.to_string());
|
||||
return Err(e);
|
||||
if let Ok(driver) = d {
|
||||
match driver.in_use().await {
|
||||
Err(err) => {
|
||||
let mut e = ERR_TIER_PERM_ERR.clone();
|
||||
e.message.push('.');
|
||||
e.message.push_str(&err.to_string());
|
||||
return Err(e);
|
||||
}
|
||||
Ok(in_use) if in_use => {
|
||||
return Err(ERR_TIER_BACKEND_NOT_EMPTY.clone());
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
Ok(in_use) if in_use => {
|
||||
return Err(ERR_TIER_BACKEND_NOT_EMPTY.clone());
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
self.tiers.remove(tier_name);
|
||||
@@ -3395,12 +3402,21 @@ impl TierConfigMgr {
|
||||
}
|
||||
|
||||
pub async fn verify(&mut self, tier_name: &str) -> std::result::Result<(), std::io::Error> {
|
||||
let driver = self.get_driver(tier_name).await.map_err(std::io::Error::other)?;
|
||||
check_warm_backend(Some(driver)).await.map_err(std::io::Error::other)
|
||||
let d = match self.get_driver(tier_name).await {
|
||||
Ok(d) => d,
|
||||
Err(err) => {
|
||||
return Err(std::io::Error::other(err));
|
||||
}
|
||||
};
|
||||
if let Err(err) = check_warm_backend(Some(d)).await {
|
||||
return Err(std::io::Error::other(err));
|
||||
} else {
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
|
||||
pub fn empty(&self) -> bool {
|
||||
self.tiers.is_empty()
|
||||
self.list_tiers().len() == 0
|
||||
}
|
||||
|
||||
pub fn tier_type(&self, tier_name: &str) -> String {
|
||||
@@ -3413,7 +3429,7 @@ impl TierConfigMgr {
|
||||
|
||||
pub fn list_tiers(&self) -> Vec<TierConfig> {
|
||||
let mut tier_cfgs = Vec::<TierConfig>::new();
|
||||
for tier in self.tiers.values() {
|
||||
for (_, tier) in self.tiers.iter() {
|
||||
let tier = tier.redacted();
|
||||
tier_cfgs.push(tier);
|
||||
}
|
||||
@@ -7119,8 +7135,7 @@ mod tests {
|
||||
let err = expect_decode_err(&encode_fixture(&wrong_hint));
|
||||
assert!(err.to_string().contains("inconsistent Wasabi type discriminators"), "{err}");
|
||||
|
||||
type WasabiPoisonField = (&'static str, fn(&mut ExternalTierS3));
|
||||
let poison_fields: [WasabiPoisonField; 6] = [
|
||||
let poison_fields: [(&str, fn(&mut ExternalTierS3)); 6] = [
|
||||
("storage_class", |s3| s3.storage_class = "GLACIER".to_string()),
|
||||
("aws_role", |s3| s3.aws_role = true),
|
||||
("web_identity_token", |s3| s3.aws_role_web_identity_token_file = "/tmp/token".to_string()),
|
||||
@@ -8241,11 +8256,7 @@ mod tests {
|
||||
peer_calls.clone(),
|
||||
Ok(PeerTierMutationState::Committed),
|
||||
)],
|
||||
TierConfigMgr::update_candidate_with_config_lock(
|
||||
&manager,
|
||||
store,
|
||||
TierCandidateMutation::Add(Box::new(tier), true),
|
||||
),
|
||||
TierConfigMgr::update_candidate_with_config_lock(&manager, store, TierCandidateMutation::Add(tier, true)),
|
||||
),
|
||||
)
|
||||
.await
|
||||
@@ -8284,7 +8295,7 @@ mod tests {
|
||||
let add = TIER_DRIVER_TEST_FACTORY.scope(
|
||||
factory,
|
||||
apply_tier_candidate_mutation(
|
||||
TierCandidateMutation::Add(Box::new(build_rustfs_tier("COLD-DEADLINE")), false),
|
||||
TierCandidateMutation::Add(build_rustfs_tier("COLD-DEADLINE"), false),
|
||||
&mut candidate,
|
||||
deadline,
|
||||
),
|
||||
@@ -9103,9 +9114,7 @@ mod tests {
|
||||
fn decode_hex_fixture(hex: &str) -> Vec<u8> {
|
||||
assert_eq!(hex.len() % 2, 0, "hex fixture must contain complete bytes");
|
||||
hex.as_bytes()
|
||||
.as_chunks::<2>()
|
||||
.0
|
||||
.iter()
|
||||
.chunks_exact(2)
|
||||
.map(|pair| {
|
||||
let pair = std::str::from_utf8(pair).expect("hex fixture should be ASCII");
|
||||
u8::from_str_radix(pair, 16).expect("hex fixture should contain only hexadecimal digits")
|
||||
@@ -11119,7 +11128,7 @@ mod tests {
|
||||
store.clone(),
|
||||
candidate,
|
||||
version,
|
||||
TierCandidateMutation::Add(Box::new(build_rustfs_tier("COLD-A")), true),
|
||||
TierCandidateMutation::Add(build_rustfs_tier("COLD-A"), true),
|
||||
update,
|
||||
None,
|
||||
)
|
||||
@@ -11716,9 +11725,8 @@ mod tests {
|
||||
assert!(merged[0].has_peer_record && merged[0].has_coordinator_record);
|
||||
}
|
||||
|
||||
let err =
|
||||
TierConfigMgr::merge_mutation_recovery_intents(std::slice::from_ref(&committed), std::slice::from_ref(&prepared))
|
||||
.expect_err("a peer committed record cannot outrun the coordinator commit order");
|
||||
let err = TierConfigMgr::merge_mutation_recovery_intents(&[committed.clone()], &[prepared.clone()])
|
||||
.expect_err("a peer committed record cannot outrun the coordinator commit order");
|
||||
assert!(err.to_string().contains("conflicting states"), "{err}");
|
||||
|
||||
let mut conflicting_identity = prepared.clone();
|
||||
@@ -13563,11 +13571,9 @@ mod tests {
|
||||
let build = tokio::spawn(async move { TierConfigMgr::acquire_operation_lease(&build_manager, cold_tier).await });
|
||||
barrier.arrived.notified().await;
|
||||
|
||||
drop(
|
||||
tokio::time::timeout(Duration::from_millis(100), manager.read())
|
||||
.await
|
||||
.expect("cold driver construction must not block manager readers"),
|
||||
);
|
||||
tokio::time::timeout(Duration::from_millis(100), manager.read())
|
||||
.await
|
||||
.expect("cold driver construction must not block manager readers");
|
||||
let tier_b = tokio::time::timeout(Duration::from_millis(100), TierConfigMgr::acquire_operation_lease(&manager, "COLD-B"))
|
||||
.await
|
||||
.expect("cold tier A construction must not block tier B")
|
||||
@@ -13770,11 +13776,9 @@ mod tests {
|
||||
let verify_manager = manager.clone();
|
||||
let verify = tokio::spawn(async move { TierConfigMgr::verify_without_manager_lock(&verify_manager, "COLD-A").await });
|
||||
started.notified().await;
|
||||
drop(
|
||||
tokio::time::timeout(Duration::from_millis(100), manager.read())
|
||||
.await
|
||||
.expect("slow verify must not hold the manager lock"),
|
||||
);
|
||||
tokio::time::timeout(Duration::from_millis(100), manager.read())
|
||||
.await
|
||||
.expect("slow verify must not hold the manager lock");
|
||||
release.add_permits(1);
|
||||
verify.await.expect("verify task should join").expect("verify should finish");
|
||||
}
|
||||
@@ -14182,11 +14186,9 @@ mod tests {
|
||||
vec!["COLD-A".to_string()]
|
||||
);
|
||||
}
|
||||
drop(
|
||||
tokio::time::timeout(Duration::from_secs(1), manager.read())
|
||||
.await
|
||||
.expect("manager reads must not wait for tier A leases"),
|
||||
);
|
||||
tokio::time::timeout(Duration::from_secs(1), manager.read())
|
||||
.await
|
||||
.expect("manager reads must not wait for tier A leases");
|
||||
let next_b = tokio::time::timeout(Duration::from_secs(1), TierConfigMgr::acquire_operation_lease(&manager, "COLD-B"))
|
||||
.await
|
||||
.expect("tier B lease acquisition must not wait for tier A")
|
||||
@@ -14619,7 +14621,7 @@ mod tests {
|
||||
"https://example-compat.invalid"
|
||||
);
|
||||
let runtime = registered_tier_driver_runtime(&manager_guard).expect("runtime sidecar should remain registered");
|
||||
assert!(!lock_unpoisoned(&runtime).generations.contains_key("COLD-A"));
|
||||
assert!(lock_unpoisoned(&runtime).generations.get("COLD-A").is_none());
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
@@ -15259,12 +15261,15 @@ mod tests {
|
||||
.filter(|object| object.bucket == bucket && object.name.starts_with(prefix))
|
||||
.cloned()
|
||||
.collect();
|
||||
objects.sort_by_key(tier_test_object_marker);
|
||||
objects.sort_by(|left, right| tier_test_object_marker(left).cmp(&tier_test_object_marker(right)));
|
||||
if marker.is_some() || version_marker.is_some() {
|
||||
let marker = (marker.unwrap_or_default(), version_marker.unwrap_or_default());
|
||||
objects.retain(|object| tier_test_object_marker(object) > marker);
|
||||
}
|
||||
let limit: usize = usize::try_from(max_keys).unwrap_or_default();
|
||||
let limit = match usize::try_from(max_keys) {
|
||||
Ok(limit) => limit,
|
||||
Err(_) => 0,
|
||||
};
|
||||
let is_truncated = objects.len() > limit;
|
||||
if is_truncated {
|
||||
objects.truncate(limit);
|
||||
@@ -15294,16 +15299,17 @@ mod tests {
|
||||
result: Self::WalkResultSender,
|
||||
opts: Self::WalkOptions,
|
||||
) -> Result<()> {
|
||||
if self.fail_reference_walk.load(Ordering::SeqCst)
|
||||
&& result
|
||||
if self.fail_reference_walk.load(Ordering::SeqCst) {
|
||||
if result
|
||||
.send(StorageObjectInfoOrErr {
|
||||
item: None,
|
||||
err: Some(Error::other("injected tier reference walk failure")),
|
||||
})
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
return Ok(());
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
let mut objects = self
|
||||
.listed_versions
|
||||
@@ -15314,7 +15320,7 @@ mod tests {
|
||||
.filter(|object| opts.include_free_versions || !object.transitioned_object.free_version)
|
||||
.cloned()
|
||||
.collect::<Vec<_>>();
|
||||
objects.sort_by_key(tier_test_object_marker);
|
||||
objects.sort_by(|left, right| tier_test_object_marker(left).cmp(&tier_test_object_marker(right)));
|
||||
if let Some(marker) = opts.marker.as_deref() {
|
||||
objects.retain(|object| object.name.as_str() > marker);
|
||||
}
|
||||
@@ -15492,18 +15498,17 @@ mod tests {
|
||||
api_view.rustfs.expect("admin RustFS payload should exist").secret_key,
|
||||
TIER_CREDENTIAL_REDACTED
|
||||
);
|
||||
{
|
||||
let observed = lock_unpoisoned(&observed);
|
||||
assert_eq!(observed.len(), 1);
|
||||
assert_eq!(
|
||||
observed[0]
|
||||
.rustfs
|
||||
.as_ref()
|
||||
.expect("backend factory should observe the RustFS payload")
|
||||
.secret_key,
|
||||
SECRET_KEY
|
||||
);
|
||||
}
|
||||
let observed = lock_unpoisoned(&observed);
|
||||
assert_eq!(observed.len(), 1);
|
||||
assert_eq!(
|
||||
observed[0]
|
||||
.rustfs
|
||||
.as_ref()
|
||||
.expect("backend factory should observe the RustFS payload")
|
||||
.secret_key,
|
||||
SECRET_KEY
|
||||
);
|
||||
drop(observed);
|
||||
|
||||
let operations = backend.op_log().await;
|
||||
assert_eq!(operations.len(), 5);
|
||||
@@ -16704,7 +16709,7 @@ mod tests {
|
||||
candidate.tiers.insert("COLD-A".to_string(), build_rustfs_tier("COLD-A"));
|
||||
candidate.tiers.insert("COLD-B".to_string(), build_rustfs_tier("COLD-B"));
|
||||
|
||||
let targets = TierCandidateMutation::Add(Box::new(build_rustfs_tier("COLD-B")), true)
|
||||
let targets = TierCandidateMutation::Add(build_rustfs_tier("COLD-B"), true)
|
||||
.affected_targets(¤t, &candidate)
|
||||
.expect("add proof should ignore unchanged durable tiers");
|
||||
assert_eq!(targets.len(), 1);
|
||||
@@ -16730,7 +16735,7 @@ mod tests {
|
||||
TierConfigMgr::update_candidate_with_config_lock(
|
||||
&manager,
|
||||
store.clone(),
|
||||
TierCandidateMutation::Add(Box::new(build_rustfs_tier("COLD-B")), true),
|
||||
TierCandidateMutation::Add(build_rustfs_tier("COLD-B"), true),
|
||||
),
|
||||
)
|
||||
.await
|
||||
@@ -16789,15 +16794,14 @@ mod tests {
|
||||
.await
|
||||
.expect("legacy nested-name Add must run the full coordinator fanout");
|
||||
|
||||
{
|
||||
let prepared_intents = lock_unpoisoned(&prepared_intents);
|
||||
assert_eq!(prepared_intents.len(), 1);
|
||||
assert_eq!(prepared_intents[0].kind, TierMutationIntentKind::Add);
|
||||
assert_eq!(prepared_intents[0].affected_targets.len(), 1);
|
||||
assert_eq!(prepared_intents[0].affected_targets[0].tier_name, "COLD-LEGACY");
|
||||
assert!(prepared_intents[0].affected_targets[0].old_backend_identity.is_none());
|
||||
assert!(prepared_intents[0].affected_targets[0].new_backend_identity.is_some());
|
||||
}
|
||||
let prepared_intents = lock_unpoisoned(&prepared_intents);
|
||||
assert_eq!(prepared_intents.len(), 1);
|
||||
assert_eq!(prepared_intents[0].kind, TierMutationIntentKind::Add);
|
||||
assert_eq!(prepared_intents[0].affected_targets.len(), 1);
|
||||
assert_eq!(prepared_intents[0].affected_targets[0].tier_name, "COLD-LEGACY");
|
||||
assert!(prepared_intents[0].affected_targets[0].old_backend_identity.is_none());
|
||||
assert!(prepared_intents[0].affected_targets[0].new_backend_identity.is_some());
|
||||
drop(prepared_intents);
|
||||
|
||||
let peer_calls = lock_unpoisoned(&peer_calls).clone();
|
||||
let prepare_index = peer_calls
|
||||
@@ -16879,7 +16883,7 @@ mod tests {
|
||||
let err = TierConfigMgr::update_candidate_with_config_lock(
|
||||
&manager,
|
||||
store.clone(),
|
||||
TierCandidateMutation::Add(Box::new(build_rustfs_tier("COLD-B")), true),
|
||||
TierCandidateMutation::Add(build_rustfs_tier("COLD-B"), true),
|
||||
)
|
||||
.await
|
||||
.expect_err("a new tier config update must wait for pending mutation recovery");
|
||||
@@ -17155,7 +17159,7 @@ mod tests {
|
||||
TierConfigMgr::update_candidate_with_config_lock(
|
||||
&update_manager,
|
||||
update_store,
|
||||
TierCandidateMutation::Add(Box::new(build_rustfs_tier("COLD-A")), true),
|
||||
TierCandidateMutation::Add(build_rustfs_tier("COLD-A"), true),
|
||||
),
|
||||
)
|
||||
.await
|
||||
@@ -17206,7 +17210,7 @@ mod tests {
|
||||
TierConfigMgr::update_candidate_with_config_lock(
|
||||
&update_manager,
|
||||
update_store,
|
||||
TierCandidateMutation::Add(Box::new(build_rustfs_tier("COLD-A")), true),
|
||||
TierCandidateMutation::Add(build_rustfs_tier("COLD-A"), true),
|
||||
),
|
||||
),
|
||||
)
|
||||
@@ -17269,7 +17273,7 @@ mod tests {
|
||||
TierConfigMgr::prevalidate_candidate_owned(
|
||||
empty_mgr(),
|
||||
None,
|
||||
TierCandidateMutation::Add(Box::new(build_rustfs_tier("COLD-A")), true),
|
||||
TierCandidateMutation::Add(build_rustfs_tier("COLD-A"), true),
|
||||
),
|
||||
)
|
||||
.await;
|
||||
@@ -17933,7 +17937,7 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn tier_add_succeeds_with_refresh_during_coordinator_commit() {
|
||||
assert_coordinator_commit_refresh_succeeds(TierCandidateMutation::Add(Box::new(build_rustfs_tier("COLD-A")), true)).await;
|
||||
assert_coordinator_commit_refresh_succeeds(TierCandidateMutation::Add(build_rustfs_tier("COLD-A"), true)).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -15,6 +15,8 @@
|
||||
#![allow(unused_variables)]
|
||||
#![allow(unused_mut)]
|
||||
#![allow(unused_assignments)]
|
||||
#![allow(unused_must_use)]
|
||||
#![allow(clippy::all)]
|
||||
|
||||
use crate::error::is_err_bucket_not_found;
|
||||
#[cfg(feature = "gcs")]
|
||||
@@ -717,7 +719,17 @@ async fn check_warm_backend_with_deadlines(
|
||||
if !matches!(cleanup_result, Ok(Ok(()))) {
|
||||
return Err(probe_cleanup_incomplete_error());
|
||||
}
|
||||
read_result?;
|
||||
if let Err(err) = read_result {
|
||||
//if is_err_bucket_not_found(&err) {
|
||||
// return Err(ERR_TIER_BUCKET_NOT_FOUND);
|
||||
//}
|
||||
/*else if is_err_signature_does_not_match(err) {
|
||||
return Err(ERR_TIER_MISSING_CREDENTIALS);
|
||||
}*/
|
||||
//else {
|
||||
return Err(err);
|
||||
//}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -747,7 +759,7 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBack
|
||||
warn!("{}", err);
|
||||
return Err(AdminError {
|
||||
code: "XRustFSAdminTierInvalidConfig".to_string(),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {err}"),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {}", err.to_string()),
|
||||
status_code: StatusCode::BAD_REQUEST,
|
||||
});
|
||||
}
|
||||
@@ -788,7 +800,7 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBack
|
||||
warn!("{}", err);
|
||||
return Err(AdminError {
|
||||
code: "XRustFSAdminTierInvalidConfig".to_string(),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {err}"),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {}", err.to_string()),
|
||||
status_code: StatusCode::BAD_REQUEST,
|
||||
});
|
||||
}
|
||||
@@ -808,7 +820,7 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBack
|
||||
warn!("{}", err);
|
||||
return Err(AdminError {
|
||||
code: "XRustFSAdminTierInvalidConfig".to_string(),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {err}"),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {}", err.to_string()),
|
||||
status_code: StatusCode::BAD_REQUEST,
|
||||
});
|
||||
}
|
||||
@@ -828,7 +840,7 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBack
|
||||
warn!("{}", err);
|
||||
return Err(AdminError {
|
||||
code: "XRustFSAdminTierInvalidConfig".to_string(),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {err}"),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {}", err.to_string()),
|
||||
status_code: StatusCode::BAD_REQUEST,
|
||||
});
|
||||
}
|
||||
@@ -848,7 +860,7 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBack
|
||||
warn!("{}", err);
|
||||
return Err(AdminError {
|
||||
code: "XRustFSAdminTierInvalidConfig".to_string(),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {err}"),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {}", err.to_string()),
|
||||
status_code: StatusCode::BAD_REQUEST,
|
||||
});
|
||||
}
|
||||
@@ -868,7 +880,7 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBack
|
||||
warn!("{}", err);
|
||||
return Err(AdminError {
|
||||
code: "XRustFSAdminTierInvalidConfig".to_string(),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {err}"),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {}", err.to_string()),
|
||||
status_code: StatusCode::BAD_REQUEST,
|
||||
});
|
||||
}
|
||||
@@ -888,7 +900,7 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBack
|
||||
warn!("{}", err);
|
||||
return Err(AdminError {
|
||||
code: "XRustFSAdminTierInvalidConfig".to_string(),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {err}"),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {}", err.to_string()),
|
||||
status_code: StatusCode::BAD_REQUEST,
|
||||
});
|
||||
}
|
||||
@@ -917,7 +929,7 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBack
|
||||
warn!("{}", err);
|
||||
return Err(AdminError {
|
||||
code: "XRustFSAdminTierInvalidConfig".to_string(),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {err}"),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {}", err.to_string()),
|
||||
status_code: StatusCode::BAD_REQUEST,
|
||||
});
|
||||
}
|
||||
@@ -937,7 +949,7 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBack
|
||||
warn!("{}", err);
|
||||
return Err(AdminError {
|
||||
code: "XRustFSAdminTierInvalidConfig".to_string(),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {err}"),
|
||||
message: format!("Unable to setup remote tier, check tier configuration: {}", err.to_string()),
|
||||
status_code: StatusCode::BAD_REQUEST,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -15,6 +15,8 @@
|
||||
#![allow(unused_variables)]
|
||||
#![allow(unused_mut)]
|
||||
#![allow(unused_assignments)]
|
||||
#![allow(unused_must_use)]
|
||||
#![allow(clippy::all)]
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
@@ -104,35 +106,39 @@ impl WarmBackendS3 {
|
||||
};
|
||||
validate_outbound_url(&u).map_err(|err| std::io::Error::other(format!("tier endpoint is not allowed: {err}")))?;
|
||||
|
||||
let has_web_identity_token_file = !conf.aws_role_web_identity_token_file.is_empty();
|
||||
let has_role_arn = !conf.aws_role_arn.is_empty();
|
||||
let has_access_key = !conf.access_key.is_empty();
|
||||
let has_secret_key = !conf.secret_key.is_empty();
|
||||
|
||||
if has_web_identity_token_file != has_role_arn {
|
||||
if conf.aws_role_web_identity_token_file == "" && conf.aws_role_arn != ""
|
||||
|| conf.aws_role_web_identity_token_file != "" && conf.aws_role_arn == ""
|
||||
{
|
||||
return Err(std::io::Error::other("both the token file and the role ARN are required"));
|
||||
} else if has_access_key != has_secret_key {
|
||||
} else if conf.access_key == "" && conf.secret_key != "" || conf.access_key != "" && conf.secret_key == "" {
|
||||
return Err(std::io::Error::other("both the access and secret keys are required"));
|
||||
} else if conf.aws_role && (has_web_identity_token_file || has_role_arn || has_access_key || has_secret_key) {
|
||||
} else if conf.aws_role
|
||||
&& (conf.aws_role_web_identity_token_file != ""
|
||||
|| conf.aws_role_arn != ""
|
||||
|| conf.access_key != ""
|
||||
|| conf.secret_key != "")
|
||||
{
|
||||
return Err(std::io::Error::other(
|
||||
"AWS Role cannot be activated with static credentials or the web identity token file",
|
||||
));
|
||||
} else if conf.bucket.is_empty() {
|
||||
} else if conf.bucket == "" {
|
||||
return Err(std::io::Error::other("no bucket name was provided"));
|
||||
}
|
||||
|
||||
let creds = if has_access_key && has_secret_key {
|
||||
let creds: Credentials<Static>;
|
||||
|
||||
if conf.access_key != "" && conf.secret_key != "" {
|
||||
//creds = Credentials::new_static_v4(conf.access_key, conf.secret_key, "");
|
||||
Credentials::new(Static(Value {
|
||||
creds = Credentials::new(Static(Value {
|
||||
access_key_id: conf.access_key.clone(),
|
||||
secret_access_key: conf.secret_key.clone(),
|
||||
session_token: "".to_string(),
|
||||
signer_type: SignatureType::SignatureV4,
|
||||
..Default::default()
|
||||
}))
|
||||
}));
|
||||
} else {
|
||||
return Err(std::io::Error::other("insufficient parameters for S3 backend authentication"));
|
||||
};
|
||||
}
|
||||
let timeouts = transition_client_timeouts_from_env();
|
||||
let opts = Options {
|
||||
creds,
|
||||
@@ -156,11 +162,11 @@ impl WarmBackendS3 {
|
||||
}
|
||||
|
||||
pub fn get_dest(&self, object: &str) -> String {
|
||||
if self.prefix.is_empty() {
|
||||
object.to_string()
|
||||
} else {
|
||||
format!("{}/{}", self.prefix, object)
|
||||
let mut dest_obj = object.to_string();
|
||||
if self.prefix != "" {
|
||||
dest_obj = format!("{}/{}", &self.prefix, object);
|
||||
}
|
||||
return dest_obj;
|
||||
}
|
||||
|
||||
pub(crate) async fn remove_with_result(&self, object: &str, rv: &str) -> Result<RemoveObjectResult, std::io::Error> {
|
||||
@@ -407,10 +413,6 @@ impl TransitionCandidateVersions {
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[allow(
|
||||
clippy::items_after_test_module,
|
||||
reason = "keep parsing tests adjacent to the helpers they cover"
|
||||
)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use rustfs_s3_client::api_s3_datatypes::{ListVersionsResult, Version};
|
||||
@@ -915,7 +917,7 @@ impl WarmBackend for WarmBackendS3 {
|
||||
.list_objects_v2(&self.bucket, &self.prefix, "", "", SLASH_SEPARATOR, 1)
|
||||
.await?;
|
||||
|
||||
Ok(!result.common_prefixes.is_empty() || !result.contents.is_empty())
|
||||
Ok(result.common_prefixes.len() > 0 || result.contents.len() > 0)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -62,6 +62,7 @@ Object keys are stored as file-system paths under each drive (`{drive}/{bucket}/
|
||||
| Behavior | RustFS | AWS S3 | Why |
|
||||
|---|---|---|---|
|
||||
| Object key with a `.` or `..` path segment, or an empty segment (`//`), such as `a//b/./c/../d` | `400 InvalidArgument` (`check_object_args` in `crates/ecstore/src/bucket/utils.rs`, mirroring MinIO `IsValidObjectPrefix`) | Accepted as an opaque key | A `..` segment would resolve to a parent directory and `.`/`//` segments would alias other keys on disk; encoding them would change the MinIO-compatible on-disk format. |
|
||||
| Directory marker (key ending in `/`, with or without a body) in a versioned bucket | Stored as the null version: `PutObject`/`HeadObject` report version id `00000000-0000-0000-0000-000000000000`, `ListObjectVersions` reports `null`, and a later PUT of the same key overwrites in place (`put_opts` in `rustfs/src/storage/options.rs`, mirroring MinIO `putOpts`: "for directory objects skip creating new versions") | A real version id per PUT, with a version history | The marker only exists to make an empty prefix listable; keeping a history for it would leave hidden versions behind every prefix delete. Replication still copies the marker as its null version (`test_bucket_replication_replicates_directory_marker_in_versioned_bucket` in `crates/e2e_test/src/replication_extension_test.rs`). |
|
||||
|
||||
## Update Rule
|
||||
|
||||
|
||||
@@ -6,7 +6,7 @@
|
||||
## What a replication PUT carries by default
|
||||
|
||||
- A plain signed body with an exact `Content-Length`. The SDK does not add a streaming trailer checksum, so the body is never wrapped in `aws-chunked` framing (rustfs#6853: a target that does not decode that framing stored the frames verbatim while RustFS recorded COMPLETED).
|
||||
- Any object-level checksum the source object was uploaded with, forwarded as its `x-amz-checksum-*` header.
|
||||
- For a single-part object, the checksum the source object was uploaded with, forwarded as its `x-amz-checksum-<algorithm>` header (the value the source verified on upload). A multipart replica is rebuilt through CreateMultipartUpload/UploadPart and carries no object-level checksum header. Managed-SSE objects forward none.
|
||||
- On a PUT that carries Object Lock parameters and no forwarded checksum: `Content-MD5` derived from the source ETag, or an SDK CRC32 checksum when the ETag is not the MD5 of the wire bytes (rustfs#7082).
|
||||
- The source ETag, mtime and version id on `x-rustfs-source-*` headers (with `x-minio-source-*` twins), and the Object Lock mode, retain-until date and legal hold of the source version when present.
|
||||
- After the PUT, the target's ETag is compared with the source ETag when both are plain single-part MD5s; a mismatch fails the replication instead of reporting a corrupted replica as COMPLETED.
|
||||
@@ -17,6 +17,7 @@
|
||||
| --- | --- | --- |
|
||||
| Rejects or mis-stores `aws-chunked` bodies (SeaweedFS 3.97) | Handled by the plain-payload default above. | Outbound target matrix, `RejectAwsChunked` mode |
|
||||
| Requires `Content-MD5` or `x-amz-checksum-*` on a PutObject with Object Lock parameters (AWS S3, MinIO, Impossible Cloud, most compatible stores) | Satisfied: a locked single PUT carries `Content-MD5` derived from the source ETag (plaintext objects whose ETag is the MD5 of the wire bytes) or an SDK CRC32 checksum (multipart-layout ETags, managed SSE, SSE-C passthrough — this one is an `aws-chunked` trailer, so a target that also rejects that framing cannot take such objects). Releases before this fix (`1.0.0-rc.5`) need `RUSTFS_REPLICATION_STREAMING_CHECKSUMS=true` as a workaround. | Outbound target matrix, `RequireChecksumWithObjectLock` mode |
|
||||
| Stores `x-amz-checksum-*` from a PutObject and returns it on `HEAD ?ChecksumMode=ENABLED` (AWS S3, Wasabi, RustFS) | Satisfied for single-part objects: the replica answers with the source's checksum. Before this fix (`1.0.0-rc.5`) the checksum left the source as `x-amz-meta-<algorithm>` user metadata and no replica carried it (rustfs/backlog#2340). | Outbound target matrix, `Checksummed` shape |
|
||||
| Mints its own version ids (AWS S3, Wasabi, Impossible Cloud) | Data lands; version-addressed convergence does not. See rustfs/backlog#2085 and `docs/operations/replication-check.md` (VersionFidelity). | `replication-check`, outbound target matrix, `MintOwnVersionIds` mode |
|
||||
| Returns an ETag that is not the content MD5 without announcing SSE | Every single-part object fails ETag verification. Set `RUSTFS_REPLICATION_REPLICA_ETAG_VERIFY=false`. | Replication status FAILED with `replica etag mismatch` |
|
||||
|
||||
@@ -24,7 +25,7 @@
|
||||
|
||||
| Variable | Default | Meaning |
|
||||
| --- | --- | --- |
|
||||
| `RUSTFS_REPLICATION_STREAMING_CHECKSUMS` | unset (plain payloads) | `true` or `1` restores SDK trailer checksums (`RequestChecksumCalculation::WhenSupported`). Every streaming upload is then `aws-chunked` with an `x-amz-trailer`; use only when every target decodes that framing. |
|
||||
| `RUSTFS_REPLICATION_STREAMING_CHECKSUMS` | unset (plain payloads) | `true` or `1` restores SDK trailer checksums (`RequestChecksumCalculation::WhenSupported`). Every streaming upload is then `aws-chunked` with an `x-amz-trailer`, except a single-part PUT that forwards the source's `x-amz-checksum-*` header, which is sent plain so the target does not receive a second algorithm; use only when every target decodes that framing. |
|
||||
| `RUSTFS_REPLICATION_REPLICA_ETAG_VERIFY` | enabled | `false` or `0` disables the post-PUT ETag comparison for targets whose 32-hex ETags are legitimately not the content MD5. |
|
||||
|
||||
Both knobs are read by the RustFS process that owns the replication target, at client build time; restart the server after changing them.
|
||||
|
||||
@@ -69,8 +69,12 @@ crates/s3-client/src/transition_api.rs|clippy::all
|
||||
crates/s3-client/src/transition_api.rs|unused_must_use
|
||||
crates/s3-client/src/transition_api.rs|unused_variables
|
||||
crates/ecstore/src/services/event_notification.rs|unused_variables
|
||||
crates/ecstore/src/services/tier/tier.rs|clippy::all
|
||||
crates/ecstore/src/services/tier/tier.rs|unused_must_use
|
||||
crates/ecstore/src/services/tier/tier.rs|unused_variables
|
||||
crates/ecstore/src/services/tier/tier_admin.rs|unused_variables
|
||||
crates/ecstore/src/services/tier/warm_backend.rs|clippy::all
|
||||
crates/ecstore/src/services/tier/warm_backend.rs|unused_must_use
|
||||
crates/ecstore/src/services/tier/warm_backend.rs|unused_variables
|
||||
crates/ecstore/src/services/tier/warm_backend_aliyun.rs|unused_variables
|
||||
crates/ecstore/src/services/tier/warm_backend_azure.rs|unused_variables
|
||||
@@ -79,5 +83,7 @@ crates/ecstore/src/services/tier/warm_backend_huaweicloud.rs|unused_variables
|
||||
crates/ecstore/src/services/tier/warm_backend_minio.rs|unused_variables
|
||||
crates/ecstore/src/services/tier/warm_backend_r2.rs|unused_variables
|
||||
crates/ecstore/src/services/tier/warm_backend_rustfs.rs|unused_variables
|
||||
crates/ecstore/src/services/tier/warm_backend_s3.rs|clippy::all
|
||||
crates/ecstore/src/services/tier/warm_backend_s3.rs|unused_must_use
|
||||
crates/ecstore/src/services/tier/warm_backend_s3.rs|unused_variables
|
||||
crates/ecstore/src/services/tier/warm_backend_tencent.rs|unused_variables
|
||||
|
||||
Reference in New Issue
Block a user