gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-08 04:58:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into feat/ilm-legacy-recovery-execution

Browse Source
This commit is contained in:
houseme
2026-09-06 23:50:46 +08:00
committed by GitHub
parent 36fd4c9fc2 a04ddc237e
commit cb7b2b2e9a
31 changed files with 2566 additions and 497 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs
+2
View File
@@ -12,6 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#[cfg(test)]
pub(crate) use rustfs_filemeta::ObjectPartInfo;
pub use rustfs_replication::{MrfOpKind, MrfReplicateEntry};
pub(crate) use rustfs_replication::{
REPLICATE_EXISTING, REPLICATE_HEAL_DELETE, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction,
crates/ecstore/src/bucket/replication/replication_object_decision_boundary.rs
+2
View File
@@ -12,6 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#[cfg(test)]
pub(crate) use rustfs_replication::ReplicationMultipartPlanError;
pub use rustfs_replication::{
MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, delete_replication_state_from_config,
delete_replication_version_id, should_schedule_delete_replication, should_use_existing_delete_replication_info,
crates/ecstore/src/bucket/replication/replication_resyncer.rs
+563 -26
View File
@@ -4656,6 +4656,58 @@ where
result
}
#[derive(Debug)]
struct MultipartReplicationReadPlan {
part_number: i32,
part_size: i64,
range: Option<HTTPRangeSpec>,
next_offset: i64,
}
fn multipart_replication_read_plan(
object_info: &ObjectInfo,
obj_opts: &ObjectOptions,
mut input: ReplicationMultipartPartInput,
stored_size: usize,
is_last: bool,
) -> std::io::Result<MultipartReplicationReadPlan> {
let empty_last_part = is_last && input.part_size == 0 && stored_size == 0;
// Raw reads address stored bytes. Only untransformed legacy parts may
// substitute their stored size for a missing logical size.
if obj_opts.raw_data_movement_read || (input.part_size == 0 && !object_info.is_compressed() && !object_info.is_encrypted()) {
input.part_size = i64::try_from(stored_size).map_err(|_| {
std::io::Error::new(std::io::ErrorKind::InvalidData, "multipart replication stored part size exceeds i64")
})?;
}
if empty_last_part {
if input.offset < 0 {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"empty multipart replication part has a negative offset",
));
}
let part_number = i32::try_from(input.part_number)
.map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidData, "multipart replication part number exceeds i32"))?;
return Ok(MultipartReplicationReadPlan {
part_number,
part_size: 0,
range: None,
next_offset: input.offset,
});
}
let plan = replication_multipart_part_plan(input).map_err(std::io::Error::other)?;
Ok(MultipartReplicationReadPlan {
part_number: plan.part_number,
part_size: plan.part_size,
range: Some(HTTPRangeSpec {
is_suffix_length: false,
start: plan.range.start,
end: plan.range.end,
}),
next_offset: plan.next_offset,
})
}
async fn replicate_multipart_parts_and_complete<S: ReplicationObjectIO>(
ctx: MultipartReplicationContext<'_, S>,
upload_id: &str,
@@ -4676,35 +4728,31 @@ async fn replicate_multipart_parts_and_complete<S: ReplicationObjectIO>(
let mut header_size = replication_put_object_header_size(&put_opts);
let mut offset: i64 = 0;
for part_info in object_info.parts.iter() {
// Ciphertext passthrough (raw read) ranges over the stored part
// bytes; decrypted reads range over the logical plaintext parts.
let part_size = if obj_opts.raw_data_movement_read {
part_info.size as i64
} else {
part_info.actual_size
};
let part_plan = replication_multipart_part_plan(ReplicationMultipartPartInput {
offset,
part_number: part_info.number,
part_size,
})
.map_err(|err| std::io::Error::other(err.to_string()))?;
let range_spec = HTTPRangeSpec {
is_suffix_length: false,
start: part_plan.range.start,
end: part_plan.range.end,
};
for (index, part_info) in object_info.parts.iter().enumerate() {
let part_plan = multipart_replication_read_plan(
object_info,
obj_opts,
ReplicationMultipartPartInput {
offset,
part_number: part_info.number,
part_size: part_info.actual_size,
},
part_info.size,
index + 1 == object_info.parts.len(),
)?;
offset = part_plan.next_offset;
let part_reader = storage
.get_object_reader(src_bucket, object, Some(range_spec), HeaderMap::new(), obj_opts)
.await
.map_err(|e| std::io::Error::other(e.to_string()))?;
let part_stream = wrap_with_bandwidth_monitor_with_header(part_reader.stream, src_bucket, arn, header_size);
let byte_stream = if let Some(range_spec) = part_plan.range {
let part_reader = storage
.get_object_reader(src_bucket, object, Some(range_spec), HeaderMap::new(), obj_opts)
.await
.map_err(|e| std::io::Error::other(e.to_string()))?;
let part_stream = wrap_with_bandwidth_monitor_with_header(part_reader.stream, src_bucket, arn, header_size);
async_read_to_bytestream(part_stream)
} else {
ByteStream::from_static(b"")
};
header_size = 0;
let byte_stream = async_read_to_bytestream(part_stream);
let object_part = cli
.put_object_part(
@@ -4760,6 +4808,173 @@ async fn replicate_multipart_parts_and_complete<S: ReplicationObjectIO>(
#[cfg(test)]
mod tests {
use super::super::replication_filemeta_boundary::ReplicateTargetDecision;
use super::super::replication_object_decision_boundary::ReplicationMultipartPlanError;
#[test]
fn multipart_read_plan_preserves_legacy_plain_part_ranges() {
const MIB: usize = 1024 * 1024;
let object_info = ObjectInfo {
etag: Some("0123456789abcdef0123456789abcdef".to_string()),
size: 6 * 1024 * 1024,
..Default::default()
};
let mut offset = 0;
for (part_number, stored_size, start, end) in [
(1, 5 * MIB, 0, 5 * 1024 * 1024 - 1),
(2, MIB, 5 * 1024 * 1024, 6 * 1024 * 1024 - 1),
] {
let plan = multipart_replication_read_plan(
&object_info,
&ObjectOptions::default(),
ReplicationMultipartPartInput {
offset,
part_number,
part_size: 0,
},
stored_size,
part_number == 2,
)
.expect("legacy plain parts must use their stored sizes");
assert_eq!(plan.part_number, i32::try_from(part_number).expect("part number fits"));
assert_eq!(plan.part_size, i64::try_from(stored_size).expect("stored size fits"));
let range = plan.range.expect("a nonempty part must read a range");
assert!(!range.is_suffix_length);
assert_eq!((range.start, range.end), (start, end));
assert_eq!(plan.next_offset, end + 1);
offset = plan.next_offset;
}
assert_eq!(offset, object_info.size);
}
#[test]
fn multipart_read_plan_distinguishes_transformed_and_raw_sizes() {
for metadata in [
HashMap::from([("x-rustfs-internal-compression".to_string(), "klauspost/compress/s2".to_string())]),
HashMap::from([("x-amz-server-side-encryption".to_string(), "AES256".to_string())]),
] {
let object_info = ObjectInfo {
user_defined: Arc::new(metadata),
..Default::default()
};
assert!(object_info.is_compressed() || object_info.is_encrypted());
for raw in [false, true] {
for actual_size in [-1, 0, 5] {
let result = multipart_replication_read_plan(
&object_info,
&ObjectOptions {
raw_data_movement_read: raw,
..Default::default()
},
ReplicationMultipartPartInput {
offset: 7,
part_number: 2,
part_size: actual_size,
},
9,
true,
);
if !raw && actual_size <= 0 {
let err = result.expect_err("transformed reads cannot substitute physical bytes for unknown plaintext");
assert!(matches!(
err.get_ref().and_then(|err| err.downcast_ref::<ReplicationMultipartPlanError>()),
Some(ReplicationMultipartPlanError::InvalidPartSize { part_size })
if *part_size == actual_size
));
} else {
let plan = result.expect("the selected representation has a known positive size");
let expected_size = if raw { 9 } else { 5 };
assert_eq!(plan.part_number, 2);
assert_eq!(plan.part_size, expected_size);
let range = plan.range.expect("a nonempty part must read a range");
assert_eq!((range.start, range.end), (7, 7 + expected_size - 1));
assert_eq!(plan.next_offset, 7 + expected_size);
}
}
}
}
}
#[test]
fn multipart_read_plan_retains_an_empty_last_part_without_advancing() {
for offset in [5 * 1024 * 1024, i64::MAX] {
for raw in [false, true] {
let plan = multipart_replication_read_plan(
&ObjectInfo::default(),
&ObjectOptions {
raw_data_movement_read: raw,
..Default::default()
},
ReplicationMultipartPartInput {
offset,
part_number: 2,
part_size: 0,
},
0,
true,
)
.expect("an empty final part needs no range read");
assert_eq!(plan.part_number, 2);
assert_eq!(plan.part_size, 0);
assert!(plan.range.is_none());
assert_eq!(plan.next_offset, offset);
}
}
}
#[test]
fn multipart_read_plan_rejects_invalid_empty_parts_and_ranges() {
for (offset, part_number, actual_size, stored_size, is_last) in [
(0, 1, 0, 0, false),
(0, 2, -1, 0, true),
(0, 2, -1, 9, true),
(-1, 2, 0, 0, true),
(0, usize::try_from(i32::MAX).expect("i32 fits usize") + 1, 0, 0, true),
(i64::MAX, 2, 1, 1, true),
(i64::MAX, 2, 2, 2, true),
] {
let err = multipart_replication_read_plan(
&ObjectInfo::default(),
&ObjectOptions::default(),
ReplicationMultipartPartInput {
offset,
part_number,
part_size: actual_size,
},
stored_size,
is_last,
)
.expect_err("invalid part metadata must not become a successful transport plan");
assert!(
err.kind() == std::io::ErrorKind::InvalidData
|| err.get_ref().is_some_and(|err| { err.is::<ReplicationMultipartPlanError>() }),
"the failure must preserve a typed metadata or planner error: {err}"
);
}
}
#[cfg(target_pointer_width = "64")]
#[test]
fn multipart_read_plan_rejects_physical_size_overflow() {
for raw in [false, true] {
let err = multipart_replication_read_plan(
&ObjectInfo::default(),
&ObjectOptions {
raw_data_movement_read: raw,
..Default::default()
},
ReplicationMultipartPartInput {
offset: 0,
part_number: 1,
part_size: 0,
},
usize::MAX,
true,
)
.expect_err("a physical size outside the range API must be rejected before casting");
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
assert_eq!(err.to_string(), "multipart replication stored part size exceeds i64");
}
}
#[test]
fn same_state_terminal_retry_uses_validate_only() {
@@ -6339,4 +6554,326 @@ mod tests {
"one target's report must not silence another's"
);
}
mod multipart_transport_tests {
use super::super::super::replication_filemeta_boundary::ObjectPartInfo;
use super::super::super::replication_storage_boundary::ObjectIO as _;
use super::*;
use bytes::Bytes;
use http_body_util::{BodyExt, Full};
use std::convert::Infallible;
#[derive(Debug)]
struct Source {
body: Bytes,
info: ObjectInfo,
ranges: StdMutex<Vec<(i64, i64)>>,
full_reads: std::sync::atomic::AtomicUsize,
}
#[async_trait::async_trait]
impl super::super::super::replication_storage_boundary::ObjectIO for Source {
type Error = Error;
type RangeSpec = HTTPRangeSpec;
type HeaderMap = HeaderMap;
type ObjectOptions = ObjectOptions;
type ObjectInfo = ObjectInfo;
type GetObjectReader = GetObjectReader;
type PutObjectReader = super::super::super::replication_storage_boundary::PutObjReader;
async fn get_object_reader(
&self,
_bucket: &str,
_object: &str,
range: Option<HTTPRangeSpec>,
_headers: HeaderMap,
opts: &ObjectOptions,
) -> Result<GetObjectReader> {
assert_eq!(
opts.version_id,
self.info.version_id.map(|id| id.to_string()),
"every read retains the selected source version"
);
if range.is_none() {
self.full_reads.fetch_add(1, Ordering::Relaxed);
return Ok(GetObjectReader {
stream: Box::new(std::io::Cursor::new(self.body.clone())),
object_info: self.info.clone(),
buffered_body: None,
body_source: Default::default(),
});
}
let range = range.expect("multipart transport must request an explicit nonempty range");
assert!(!range.is_suffix_length);
assert!(range.start <= range.end, "empty parts must not issue an inverted range");
self.ranges.lock().expect("range journal lock").push((range.start, range.end));
let start = usize::try_from(range.start).expect("nonnegative start");
let end = usize::try_from(range.end).expect("nonnegative end");
let body = self.body.slice(start..=end);
Ok(GetObjectReader {
stream: Box::new(std::io::Cursor::new(body)),
object_info: self.info.clone(),
buffered_body: None,
body_source: Default::default(),
})
}
async fn put_object(
&self,
_bucket: &str,
_object: &str,
_data: &mut Self::PutObjectReader,
_opts: &ObjectOptions,
) -> Result<ObjectInfo> {
panic!("replication must not overwrite its source")
}
}
#[derive(Debug)]
struct RequestRecord {
method: http::Method,
query: HashMap<String, String>,
headers: HeaderMap,
body: Bytes,
}
#[tokio::test]
async fn multipart_transport_preserves_legacy_zero_actual_sizes() {
run_transport(4096, None).await;
}
#[tokio::test]
async fn multipart_transport_uploads_an_empty_last_part_without_reading_a_range() {
run_transport(0, None).await;
}
#[tokio::test]
async fn multipart_transport_preserves_transformed_unknown_nonempty_parts() {
for unknown_part in [(0, 0), (1, 0), (0, -1), (1, -1)] {
run_transport(4096, Some(unknown_part)).await;
}
}
#[tokio::test]
async fn multipart_transport_preserves_transformed_empty_tail() {
run_transport(0, Some((1, 0))).await;
}
async fn run_transport(tail_size: usize, unknown_part: Option<(usize, i64)>) {
const FIRST_SIZE: usize = 5 * 1024 * 1024;
let body = Bytes::from([vec![0x35; FIRST_SIZE], vec![0xa7; tail_size]].concat());
let etag = faster_hex::hex_string(rustfs_utils::hash::HashAlgorithm::Md5.hash_encode(&body).as_ref());
let source = Arc::new(Source {
info: ObjectInfo {
size: i64::try_from(body.len() + if unknown_part.is_some() { 16 } else { 0 }).expect("stored size"),
actual_size: i64::try_from(body.len()).expect("body size"),
etag: Some(etag.clone()),
version_id: Some(Uuid::new_v4()),
user_defined: Arc::new(if unknown_part.is_some() {
HashMap::from([("x-amz-server-side-encryption".to_string(), "AES256".to_string())])
} else {
HashMap::new()
}),
parts: Arc::new(vec![
ObjectPartInfo {
number: 1,
size: FIRST_SIZE + if unknown_part.is_some() { 8 } else { 0 },
actual_size: if let Some((0, size)) = unknown_part {
size
} else if unknown_part.is_some() || tail_size == 0 {
i64::try_from(FIRST_SIZE).expect("first part size")
} else {
0
},
..Default::default()
},
ObjectPartInfo {
number: 2,
size: tail_size + if unknown_part.is_some() { 8 } else { 0 },
actual_size: if let Some((1, size)) = unknown_part {
size
} else if unknown_part.is_some() {
i64::try_from(tail_size).expect("tail logical size")
} else {
0
},
..Default::default()
},
]),
..Default::default()
},
body: body.clone(),
ranges: StdMutex::new(Vec::new()),
full_reads: std::sync::atomic::AtomicUsize::new(0),
});
let journal = Arc::new(StdMutex::new(Vec::<RequestRecord>::new()));
let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
.await
.expect("bind multipart target");
let endpoint = format!("http://{}", listener.local_addr().expect("multipart target address"));
let server_journal = journal.clone();
let server = tokio::spawn(async move {
let mut connections = JoinSet::new();
loop {
let (stream, _) = listener.accept().await.expect("accept multipart request");
let journal = server_journal.clone();
connections.spawn(async move {
let service = hyper::service::service_fn(move |request: hyper::Request<hyper::body::Incoming>| {
let journal = journal.clone();
async move {
let (request, body) = request.into_parts();
let query: HashMap<String, String> = url::form_urlencoded::parse(
request.uri.query().unwrap_or_default().as_bytes(),
).into_owned().collect();
let body = body.collect().await.expect("read complete multipart request body").to_bytes();
let response = if request.method == http::Method::POST && query.contains_key("uploads") {
"<InitiateMultipartUploadResult><Bucket>target-bucket</Bucket><Key>object</Key><UploadId>upload-1</UploadId></InitiateMultipartUploadResult>"
} else if request.method == http::Method::PUT {
""
} else if request.method == http::Method::POST && query.contains_key("uploadId") {
"<CompleteMultipartUploadResult><Location>http://localhost/object</Location><Bucket>target-bucket</Bucket><Key>object</Key><ETag>&quot;target-2&quot;</ETag></CompleteMultipartUploadResult>"
} else if request.method == http::Method::DELETE && query.contains_key("uploadId") {
""
} else {
panic!("unexpected multipart request: {} {}", request.method, request.uri)
};
let response_etag = if request.method == http::Method::PUT && !query.contains_key("partNumber") {
format!("\"{}\"", faster_hex::hex_string(rustfs_utils::hash::HashAlgorithm::Md5.hash_encode(&body).as_ref()))
} else {
"\"uploaded-part\"".to_string()
};
journal.lock().expect("request journal lock").push(RequestRecord {
method: request.method, query, headers: request.headers, body,
});
Ok::<_, Infallible>(hyper::Response::builder()
.header("content-type", "application/xml")
.header("etag", response_etag)
.body(Full::new(Bytes::from_static(response.as_bytes())))
.expect("multipart response"))
}
});
hyper::server::conn::http1::Builder::new()
.serve_connection(hyper_util::rt::TokioIo::new(stream), service)
.await.expect("serve multipart connection");
});
}
});
let mut target = test_target_client(endpoint);
let config = target
.client
.config()
.to_builder()
.request_checksum_calculation(aws_sdk_s3::config::RequestChecksumCalculation::WhenRequired)
.force_path_style(true)
.build();
Arc::get_mut(&mut target).expect("unshared test target").client = Arc::new(aws_sdk_s3::Client::from_conf(config));
let (put_opts, is_multipart) = replication_put_object_options("STANDARD", &source.info).expect("replication options");
let opts = ObjectOptions {
version_id: source.info.version_id.map(|id| id.to_string()),
..Default::default()
};
let reader = source
.get_object_reader("source", "object", None, HeaderMap::new(), &opts)
.await
.expect("open the existing full-object stream");
let result = tokio::time::timeout(
std::time::Duration::from_secs(30),
replicate_all_payload_to_target(
ReplicateAllPayloadContext {
storage: &source,
tgt_client: &target,
bucket: "source",
object: "object",
object_info: &source.info,
obj_opts: &opts,
arn: &target.arn,
transfer_size: i64::try_from(body.len()).expect("plaintext size"),
is_multipart,
put_opts,
},
reader,
),
)
.await;
server.abort();
assert!(server.await.expect_err("fixture server is stopped").is_cancelled());
if let Some(error) = result.expect("replication must finish") {
panic!("legacy parts must replicate successfully: {error}");
}
assert_eq!(
source.full_reads.load(Ordering::Relaxed),
1,
"reuse the initial full stream without an extra read"
);
if unknown_part.is_some() {
let requests = journal.lock().expect("request journal lock");
assert_eq!(requests.len(), 1, "unknown transformed boundaries retain one streaming PUT");
let request = &requests[0];
assert_eq!(request.method, http::Method::PUT);
let source_version = source.info.version_id.map(|id| id.to_string()).expect("versioned fixture");
assert_eq!(
request.query,
HashMap::from([
("x-id".to_string(), "PutObject".to_string()),
("versionId".to_string(), source_version.clone()),
]),
"single PUT carries only the SDK operation query and the source versionId the target must reuse"
);
assert_eq!(request.body, body, "single PUT includes every byte of both source parts");
assert_eq!(
request.headers.get("content-length").expect("body length"),
body.len().to_string().as_str()
);
assert_eq!(
rustfs_utils::http::get_header(&request.headers, rustfs_utils::http::SUFFIX_SOURCE_ETAG).as_deref(),
Some(etag.as_str())
);
assert_eq!(
rustfs_utils::http::get_header(&request.headers, rustfs_utils::http::SUFFIX_SOURCE_VERSION_ID)
.map(|value| value.into_owned()),
Some(source_version),
"single PUT preserves the selected source version"
);
assert!(
source.ranges.lock().expect("range journal lock").is_empty(),
"unknown logical boundaries must not issue guessed ranges"
);
return;
}
let requests = journal.lock().expect("request journal lock");
assert_eq!(requests.len(), 4, "initiate, two upload parts, and complete without retries");
assert!(requests[0].query.contains_key("uploads"));
for (index, expected) in [(1, body.slice(..FIRST_SIZE)), (2, body.slice(FIRST_SIZE..))] {
assert_eq!(requests[index].method, http::Method::PUT);
assert_eq!(requests[index].query.get("partNumber"), Some(&index.to_string()));
assert_eq!(requests[index].body, expected, "upload part contains the exact source range");
assert_eq!(
requests[index].headers.get("content-length").expect("part content length"),
expected.len().to_string().as_str()
);
}
let complete = &requests[3];
assert_eq!(complete.method, http::Method::POST);
assert_eq!(
rustfs_utils::http::get_header(&complete.headers, rustfs_utils::http::SUFFIX_SOURCE_ETAG).as_deref(),
Some(etag.as_str())
);
let complete_xml = std::str::from_utf8(&complete.body).expect("complete XML");
assert_eq!(
complete_xml.matches("<Part>").count(),
2,
"the empty final part must remain in the completion list"
);
assert!(complete_xml.contains("<PartNumber>1</PartNumber>"));
assert!(complete_xml.contains("<PartNumber>2</PartNumber>"));
let mut expected_ranges = vec![(0, i64::try_from(FIRST_SIZE - 1).expect("first end"))];
if tail_size > 0 {
expected_ranges.push((
i64::try_from(FIRST_SIZE).expect("tail start"),
i64::try_from(body.len() - 1).expect("tail end"),
));
}
assert_eq!(*source.ranges.lock().expect("range journal lock"), expected_ranges);
}
}
}
crates/ecstore/src/bucket/replication/replication_target_boundary.rs
+166 -6
View File
@@ -248,7 +248,16 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
meta.insert(AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "aws:kms".to_string());
}
let mut is_multipart = object_info.is_multipart();
// Older transformed objects can have physical parts without logical part
// lengths. Keep their existing whole-object transport: physical sizes are
// not plaintext boundaries for a multipart replication read.
let legacy_single_put = object_info.etag.as_deref().is_none_or(|etag| etag.len() == 32);
let base_is_multipart = object_info.is_multipart()
&& !(legacy_single_put
&& object_info.parts.len() > 1
&& (object_info.is_compressed() || object_info.is_encrypted())
&& object_info.parts.iter().any(|part| part.actual_size <= 0));
let mut is_multipart = base_is_multipart;
if let Some(checksum_data) = &object_info.checksum
&& !checksum_data.is_empty()
@@ -259,8 +268,8 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
} else if object_info.is_encrypted() {
// Encrypted checksums cannot be exposed as plaintext headers, and
// decrypt_checksums reports is_multipart=false for them (a value
// the response path relies on). Keep the object's own multipart
// flag so encrypted objects stay on the multipart route.
// the response path relies on). Keep the transport selected from
// the object's layout and readable part boundaries.
} else {
let (checksum_meta, checksum_record_is_multipart) = object_info.decrypt_checksums(0, &HeaderMap::new())?;
// The checksum record describes how the *checksum* is composed,
@@ -268,9 +277,9 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
// MULTIPART flag even on a multipart upload, so trusting it here
// routed a 768-part object through a single PutObject and the
// target rejected the 6 GiB body with EntityTooLarge
// (rustfs#6825). The object's own shape is the authority: the
// (rustfs#6825). The usable part layout is the authority: the
// record may only add multipart-ness, never take it away.
is_multipart = object_info.is_multipart() || checksum_record_is_multipart;
is_multipart = base_is_multipart || checksum_record_is_multipart;
for (key, value) in checksum_meta.iter() {
if key != AMZ_CHECKSUM_TYPE {
@@ -278,7 +287,7 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
}
}
if !object_info.is_multipart()
if !base_is_multipart
&& checksum_meta
.get(AMZ_CHECKSUM_TYPE)
.is_some_and(|value| value == AMZ_CHECKSUM_TYPE_FULL_OBJECT)
@@ -516,6 +525,7 @@ fn is_standard_header(key: &str) -> bool {
#[cfg(test)]
mod tests {
use super::super::replication_filemeta_boundary::ObjectPartInfo;
use super::*;
use aws_smithy_types::DateTime;
use rustfs_replication::content_matches_by_etag;
@@ -550,6 +560,109 @@ mod tests {
checksum.to_bytes(&combined)
}
fn replication_route_metadata() -> [(&'static str, Arc<HashMap<String, String>>); 4] {
let mut compressed = HashMap::new();
rustfs_utils::http::insert_str(&mut compressed, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
[
("plain", Arc::new(HashMap::new())),
("compressed", Arc::new(compressed)),
(
"encrypted",
Arc::new(HashMap::from([(AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "AES256".to_string())])),
),
(
"ssec",
Arc::new(HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())])),
),
]
}
fn replication_route_object(
etag: Option<&str>,
actual_sizes: [i64; 3],
metadata: Arc<HashMap<String, String>>,
) -> ObjectInfo {
ObjectInfo {
etag: etag.map(str::to_string),
size: 48,
actual_size: 12,
user_defined: metadata,
parts: Arc::new(
actual_sizes
.into_iter()
.enumerate()
.map(|(index, actual_size)| ObjectPartInfo {
number: index + 1,
size: 16,
actual_size,
..Default::default()
})
.collect(),
),
..Default::default()
}
}
#[test]
fn legacy_transformed_single_put_parts_keep_the_previous_replication_route() {
let [_, (_, compressed), (_, encrypted), (_, ssec)] = replication_route_metadata();
let cases = [
(
"compressed middle zero",
compressed.clone(),
Some("0123456789abcdef0123456789abcdef"),
[4, 0, 4],
),
("compressed tail unknown", compressed, None, [4, 4, -1]),
(
"encrypted middle unknown",
encrypted,
Some("gggggggggggggggggggggggggggggggg"),
[4, -1, 4],
),
("ssec tail zero", ssec.clone(), None, [4, 4, 0]),
("ssec middle unknown", ssec, Some("gggggggggggggggggggggggggggggggg"), [4, -1, 4]),
];
for (name, metadata, etag, actual_sizes) in cases {
for checksum in [None, Some(full_object_multipart_checksum_record())] {
let mut object_info = replication_route_object(etag, actual_sizes, metadata.clone());
object_info.checksum = checksum;
assert!(object_info.is_multipart(), "{name}: physical parts remain visible to metadata APIs");
assert!(object_info.is_compressed() || object_info.is_encrypted());
let (options, is_multipart) =
replication_put_object_options("STANDARD", &object_info).expect("legacy transformed put options");
assert!(
!is_multipart,
"{name}: unknown logical part sizes must preserve the old whole-object route"
);
assert_eq!(options.internal.source_etag, etag.unwrap_or_default());
if metadata.contains_key(SSEC_ALGORITHM_HEADER) {
assert_eq!(
get_header_map(&options.user_metadata, SUFFIX_REPLICATION_SSEC_CRC).is_some(),
object_info.checksum.is_some(),
"SSE-C checksums retain their raw passthrough transport"
);
}
}
}
}
#[test]
fn positive_part_sizes_and_legacy_multipart_etags_keep_the_replication_route() {
for (name, metadata) in replication_route_metadata() {
for (etag, actual_sizes) in [
("0123456789abcdef0123456789abcdef", [4, 4, 4]),
("0123456789abcdef0123456789abcdef-3", [4, 0, -1]),
] {
let mut object_info = replication_route_object(Some(etag), actual_sizes, metadata.clone());
object_info.checksum = Some(full_object_multipart_checksum_record());
let (_, is_multipart) = replication_put_object_options("STANDARD", &object_info).expect("multipart put options");
assert!(is_multipart, "{name}/{etag}: usable sizes and old multipart ETags must retain MPU");
}
}
}
#[test]
fn multipart_object_with_full_object_checksum_keeps_the_multipart_route() {
// rustfs#6825: a 768-part upload was replicated with a single
@@ -582,6 +695,36 @@ mod tests {
);
}
#[test]
fn stored_multipart_parts_keep_the_replication_route_without_a_multipart_etag() {
for etag in [Some("0123456789abcdef0123456789abcdef"), None] {
for checksum in [None, Some(full_object_multipart_checksum_record())] {
let object_info = ObjectInfo {
etag: etag.map(str::to_string),
checksum,
parts: Arc::new(
(1..=2)
.map(|number| ObjectPartInfo {
number,
..Default::default()
})
.collect(),
),
..Default::default()
};
let (options, is_multipart) =
replication_put_object_options("STANDARD", &object_info).expect("build put options");
assert!(
is_multipart,
"stored parts must retain multipart routing: etag={etag:?}, checksum={:?}",
object_info.checksum
);
assert_eq!(options.internal.source_etag, etag.unwrap_or_default());
}
}
}
#[test]
fn checksum_record_never_changes_the_transport_a_single_part_object_needs() {
// The mirror of the rustfs#6825 guard: an object stored as one PUT
@@ -592,6 +735,10 @@ mod tests {
let object_info = ObjectInfo {
etag: Some("0123456789abcdef0123456789abcdef".to_string()),
checksum: Some(checksum.to_bytes(&[])),
parts: Arc::new(vec![ObjectPartInfo {
number: 1,
..Default::default()
}]),
..Default::default()
};
@@ -628,6 +775,19 @@ mod tests {
let (_, is_multipart) = replication_put_object_options("STANDARD", &object_info).expect("build put options");
assert!(is_multipart, "a composite-checksum multipart object must stay on the multipart transport");
for (name, metadata) in replication_route_metadata() {
let mut legacy = replication_route_object(Some("0123456789abcdef0123456789abcdef"), [4, 0, 4], metadata);
legacy.checksum = Some(checksum.to_bytes(&combined));
let (_, record_is_multipart) = legacy.decrypt_checksums(0, &HeaderMap::new()).expect("decode checksum");
let (_, is_multipart) = replication_put_object_options("STANDARD", &legacy).expect("legacy checksum put options");
if legacy.is_encrypted() {
assert!(!is_multipart, "{name}: encrypted checksum records must not change the old transport");
} else {
assert!(record_is_multipart, "the composite checksum must carry its own multipart signal");
assert!(is_multipart, "{name}: a composite record can still promote the legacy route to MPU");
}
}
}
#[test]
crates/ecstore/src/data_movement/mod.rs
+2 -2
View File
@@ -2673,7 +2673,7 @@ mod tests {
]),
..Default::default()
};
assert!(!object_info.is_multipart());
assert!(object_info.is_multipart());
assert!(should_use_multipart_data_movement(&object_info, false));
let single_nonstandard_part = ObjectInfo {
@@ -3050,7 +3050,7 @@ mod tests {
..Default::default()
};
assert!(!object_info.is_multipart());
assert!(object_info.is_multipart());
assert!(object_info.parts.iter().any(|part| part.checksums.is_some()));
let opts = data_movement_put_object_opts(&object_info, 0);
assert!(!rustfs_utils::http::contains_key_str(&opts.user_defined, SUFFIX_PART_CHECKSUMS));
crates/ecstore/src/object_api/readers.rs
+170
View File
@@ -2278,6 +2278,121 @@ mod tests {
assert_eq!(read, fixture.plaintext, "SSE-C + compression full GET must reassemble all parts");
}
#[tokio::test]
async fn multipart_empty_tail_full_reads_preserve_plaintext() {
let key = [0x6Eu8; 32];
let part_sizes = [5 * 1024 * 1024, 0];
let encrypted = build_legacy_ssec_multipart_fixture(key, &part_sizes).await;
for (kind, mut fixture, headers) in [
(
"encrypted",
CompressedMultipartFixture {
object_info: encrypted.object_info,
stored: encrypted.ciphertext,
plaintext: encrypted.plaintext,
},
ssec_headers_from_key(key),
),
("compressed", compressed_multipart_fixture(&part_sizes).await, HeaderMap::new()),
(
"compressed and encrypted",
compressed_encrypted_multipart_fixture(key, &part_sizes).await,
ssec_headers_from_key(key),
),
] {
fixture.object_info.etag = Some(faster_hex::hex_string(Md5::digest(&fixture.plaintext).as_ref()));
assert_eq!(fixture.object_info.etag.as_ref().expect("source ETag").len(), 32);
assert_eq!(fixture.object_info.parts.len(), 2);
let tail = &fixture.object_info.parts[1];
assert_eq!(tail.actual_size, 0, "{kind}: final part has no plaintext");
if kind == "compressed" {
assert_eq!(tail.size, 0, "unpadded compression emits no bytes for an empty part");
} else {
assert!(tail.size > 0, "{kind}: the empty part still has a stored frame");
}
let stored_size = i64::try_from(fixture.stored.len()).expect("fixture size fits i64");
let (mut reader, offset, length) = GetObjectReader::new(
Box::new(Cursor::new(fixture.stored)),
None,
&fixture.object_info,
&ObjectOptions::default(),
&headers,
)
.await
.expect("full transformed read must include the empty tail");
assert_eq!((offset, length), (0, stored_size), "{kind}: full read includes all stored parts");
let mut body = Vec::new();
reader
.stream
.read_to_end(&mut body)
.await
.expect("read through the complete decoder EOF");
assert_eq!(body, fixture.plaintext, "{kind}: no plaintext is added or lost by the empty tail");
}
}
#[tokio::test]
async fn multipart_empty_tail_full_read_authenticates_v2_final_frame() {
let key = [0x6Eu8; 32];
let plaintext = legacy_fixture_part_plaintext(1, 5 * 1024 * 1024);
let mut ciphertext = Vec::new();
let mut parts = Vec::new();
for (number, body) in [(1, plaintext.as_slice()), (2, b"".as_slice())] {
let start = ciphertext.len();
rustfs_rio::EncryptReader::new_multipart_v2(Cursor::new(body), key, LEGACY_FIXTURE_BASE_NONCE, number)
.read_to_end(&mut ciphertext)
.await
.expect("encrypt a v2 fixture part with an authenticated final frame");
parts.push(ObjectPartInfo {
number,
size: ciphertext.len() - start,
actual_size: i64::try_from(body.len()).expect("fixture plaintext size fits"),
..Default::default()
});
}
let tail_start = parts[0].size;
assert_eq!(parts[1].actual_size, 0);
assert!(parts[1].size > 8, "the empty final frame carries more than an END marker");
let object_info = ObjectInfo {
bucket: "bucket".to_string(),
name: "v2-empty-tail".to_string(),
size: i64::try_from(ciphertext.len()).expect("fixture ciphertext size fits"),
etag: Some(faster_hex::hex_string(Md5::digest(&plaintext).as_ref())),
parts: Arc::new(parts),
user_defined: Arc::new(legacy_ssec_multipart_metadata(key, plaintext.len())),
..Default::default()
};
for corrupt_tail in [false, true] {
let mut stored = ciphertext.clone();
if corrupt_tail {
// The v2 header is authenticated associated data, including
// the header of a final frame containing zero plaintext.
stored[tail_start + 5] ^= 1;
}
let (mut reader, offset, length) = GetObjectReader::new(
Box::new(Cursor::new(stored)),
None,
&object_info,
&ObjectOptions::default(),
&ssec_headers_from_key(key),
)
.await
.expect("construct the full reader before consuming the final frame");
assert_eq!((offset, length), (0, object_info.size));
let result = tokio::io::copy(&mut reader.stream, &mut tokio::io::sink()).await;
if corrupt_tail {
let err = result.expect_err("EOF must authenticate the empty final frame after all plaintext is returned");
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
assert_eq!(err.to_string(), "v2 encrypted frame failed authentication");
} else {
assert_eq!(
result.expect("valid empty final frame must reach EOF"),
u64::try_from(plaintext.len()).expect("plaintext length fits")
);
}
}
}
#[tokio::test]
async fn compressed_encrypted_multipart_range_crosses_part_boundary() {
let key_bytes = [0x6Eu8; 32];
@@ -3656,6 +3771,61 @@ mod tests {
.await;
}
#[tokio::test]
async fn multipart_full_read_preserves_legacy_zero_and_negative_part_sizes() {
let key = [0x77; 32];
let part_sizes = [5 * 1024 * 1024, 1024 * 1024];
let encrypted = build_legacy_ssec_multipart_fixture(key, &part_sizes).await;
// The encrypted case supplies the fixture key explicitly. This covers
// full decrypted reads, not managed-key acquisition.
for (kind, fixture, headers) in [
("compressed", compressed_multipart_fixture(&part_sizes).await, HeaderMap::new()),
(
"encrypted with supplied key",
CompressedMultipartFixture {
object_info: encrypted.object_info,
stored: encrypted.ciphertext,
plaintext: encrypted.plaintext,
},
ssec_headers_from_key(key),
),
] {
let source_etag = faster_hex::hex_string(Md5::digest(&fixture.plaintext).as_ref());
assert_eq!(source_etag.len(), 32);
assert_eq!(fixture.plaintext.len(), 6 * 1024 * 1024);
for part_index in 0..part_sizes.len() {
assert!(fixture.object_info.parts[part_index].actual_size > 0, "the selected part is nonempty");
for actual_size in [0, -1] {
let mut object_info = fixture.object_info.clone();
object_info.etag = Some(source_etag.clone());
Arc::make_mut(&mut object_info.parts)[part_index].actual_size = actual_size;
let (mut reader, offset, length) = GetObjectReader::new(
Box::new(Cursor::new(fixture.stored.clone())),
None,
&object_info,
&ObjectOptions::default(),
&headers,
)
.await
.expect("the authoritative total size must keep full legacy reads available");
assert_eq!(offset, 0);
assert_eq!(length, i64::try_from(fixture.stored.len()).expect("stored size fits"));
let mut body = Vec::new();
reader
.stream
.read_to_end(&mut body)
.await
.expect("full read must reach EOF despite an unspecified per-part logical size");
assert_eq!(
body, fixture.plaintext,
"{kind}: part {part_index} with actual_size={actual_size} must not lose readable data"
);
assert_eq!(reader.object_info.etag.as_deref(), Some(source_etag.as_str()));
}
}
}
}
/// The physical part sizes must add up to `oi.size` for a seek to be safe;
/// inconsistent metadata must fall back to the previous full-object read
/// instead of scheduling an erasure read past the object end.
crates/ecstore/src/object_api/types.rs
+30 -1
View File
@@ -1597,7 +1597,7 @@ impl ObjectInfo {
}
pub fn is_multipart(&self) -> bool {
self.etag.as_ref().is_some_and(|v| v.len() != 32)
self.parts.len() > 1 || self.etag.as_ref().is_some_and(|v| v.len() != 32)
}
pub fn is_encrypted(&self) -> bool {
@@ -2235,6 +2235,35 @@ mod tests {
}
use rustfs_filemeta::{FileInfo, FileMeta, MetaCacheEntry, TRANSITION_COMPLETE};
#[test]
fn multipart_identity_uses_stored_parts_and_preserves_the_etag_fallback() {
let plain_etag = "0123456789abcdef0123456789abcdef";
let multipart_etag = "0123456789abcdef0123456789abcdef-1";
for (case, part_count, etag, expected) in [
("preserved source ETag", 2, Some(plain_etag), true),
("missing ETag", 2, None, true),
("ordinary PUT", 1, Some(plain_etag), false),
("ordinary PUT without ETag", 1, None, false),
("single-part MPU", 1, Some(multipart_etag), true),
("legacy MPU without parts", 0, Some(multipart_etag), true),
] {
let object = ObjectInfo {
etag: etag.map(str::to_string),
parts: Arc::new(
(1..=part_count)
.map(|number| ObjectPartInfo {
number,
..Default::default()
})
.collect(),
),
..Default::default()
};
assert_eq!(object.is_multipart(), expected, "{case}");
}
}
fn inline_fast_path_object(size: i64, versioned: bool) -> ObjectInfo {
ObjectInfo {
size,
crates/ecstore/src/store/init.rs
+1 -1
View File
@@ -4380,7 +4380,7 @@ mod tests {
}
retry_source_info.parts = Arc::new(retry_source_parts);
assert_eq!(retry_source_info.etag.as_deref(), Some(retry_object_etag.as_str()));
assert!(!retry_source_info.is_multipart());
assert!(retry_source_info.is_multipart());
assert!(retry_source_info.parts.iter().all(|part| part.checksums.is_some()));
assert_eq!(retry_source_info.checksum.as_deref(), Some(retry_object_checksum_bytes.as_ref()));
assert!(