mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-27 15:37:02 +00:00
fix(ecstore): handle ChecksumNone in >128 MiB ILM transitions (#4831)
* fix(ecstore): treat ChecksumNone as unset so >128 MiB ILM transitions succeed ILM transition of any object larger than 128 MiB to a RustFS-native tier (rustfs/minio/aliyun/tencent/r2/azure/huaweicloud/s3 backends that use the built-in TransitionClient) failed with "unsupported checksum type", while objects <=128 MiB transitioned fine. Root cause: `ChecksumMode::is_set()` reported `ChecksumNone` as a configured checksum. `ChecksumNone` is the zeroth enum variant, so it occupies bit 0 of the EnumSet repr and the `len() == 1` check treated "no checksum" as set. The 128 MiB boundary is the warm backend's `MIN_PART_SIZE`, which selects a single PUT (<=128 MiB) versus a multipart PUT (>128 MiB). On the multipart path, `put_object_multipart_stream_optional_checksum` saw `checksum.is_set() == true`, disabled the Content-MD5 branch, and called `ChecksumNone.hasher()`, which returns the "unsupported checksum type" error. The single-PUT path hit the same misjudgement but never calls `hasher()`, so it silently succeeded (without a checksum), which is why only >128 MiB objects failed. Fix: - `is_set()` returns false for `ChecksumNone` (and the bare `ChecksumFullObject` flag, which has no base algorithm). This is the sole callers' intended meaning: a concrete algorithm with a real hasher is selected. - Defense in depth: guard the multipart checksum branch on `auto_checksum.is_set()` so an unset mode uploads the part without a per-part checksum header instead of hard-failing in `hasher()`. Only the TransitionClient consumes this `ChecksumMode::is_set()`; the server-side data path uses the unrelated `rustfs_rio::ChecksumType`. Tests: is_set()/set_default semantics, hasher parity for every set mode, and a `build_transition_put_options` invariant (checksum unset + Content-MD5 on). Refs: rustfs/rustfs#4811, rustfs/backlog#1267 Co-Authored-By: heihutu <heihutu@gmail.com> * fix(ecstore): read exactly one part per multipart chunk in transition uploads Second defect behind the >128 MiB ILM transition failure (rustfs/rustfs#4811), uncovered while verifying the checksum fix. `put_object_multipart_stream_optional_checksum` read each part with `read_all()` / `to_vec()`, which drained the entire source into the first part and left every later part empty. Any multipart upload of a streamed (`ObjectBody`) source was therefore malformed. Objects <=128 MiB take the single-part path and were unaffected; a 128 MiB + 1 byte object splits into a 128 MiB part plus a 1 byte part, so the first part received the whole object and its declared Content-Length (part_size) did not match the body. Verified empirically: `optimal_part_info(128 MiB + 1, 128 MiB)` yields 2 parts, and `GetObjectReader::read_all()` on part 1 returns the full 134217729 bytes, leaving 0 for part 2. Fix: - Add `read_multipart_part`, which reads exactly the requested part size (or less at EOF) and advances the reader, for both `Body` (in-memory) and `ObjectBody` (streamed) sources. - Upload each part with the bytes actually read (`length`) as its size, and account uploaded size by actual bytes, so a short read is detected instead of masked. The concurrent (`put_object_multipart_stream_parallel`) and SigV2 (`put_object_multipart`) paths share the same `read_all()` pattern but are not exercised by transition; left untouched here and noted for follow-up. Tests: `read_multipart_part` splits a 250-byte source into [100, 100, 50] for both streamed and in-memory bodies, consumes the source fully, and stops at EOF without overrun. Refs: rustfs/rustfs#4811, rustfs/backlog#1267 Co-Authored-By: heihutu <heihutu@gmail.com> * fix(ecstore): complete the >128 MiB ILM transition multipart client Docker end-to-end reproduction of rustfs/rustfs#4811 (two RustFS tiers, a 128 MiB + 1 byte object, zero-day transition) surfaced four more defects on the multipart transition path, each masked by the previous one. With the checksum and part-splitting fixes in place the transition now failed later and later, and finally produced a 0-byte object with no error at all. Fixed together: - initiate_multipart_upload discarded the CreateMultipartUpload response and returned an empty UploadId, so the first UploadPart failed with "UploadID cannot be empty". Parse the response XML (InitiateMultipartUploadResult now derives Deserialize with PascalCase). - Content-MD5 / x-amz-checksum-* were encoded with URL-safe, unpadded base64, which the remote rejected as "Invalid content MD5: Base64Error". Add base64_encode_standard and use it for those outbound header values. - PutObjectOptions::default() set legalhold to OFF, so header() attached x-amz-object-lock-legal-hold to every request and CompleteMultipartUpload was rejected with "does not accept object lock or governance bypass headers". Default to an empty (unset) status. - CompleteMultipartUpload / CompletePart had no serde renames, so the request body used Rust field names (<parts>/<part_num>/<etag>). The remote parsed zero <Part> elements and completed a 0-byte object while returning 200. Emit S3 element names (<Part>/<PartNumber>/<ETag>) and skip empty checksum fields. Verified end-to-end: a 128 MiB + 1 byte object now transitions to the remote tier and reads back (transparently restored) byte-for-byte identical (sha256 match), with none of the four prior errors in the logs. Refs: rustfs/rustfs#4811, rustfs/backlog#1267 Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
@@ -25,6 +25,7 @@ use std::io::Error;
|
||||
use std::sync::RwLock;
|
||||
use std::{collections::HashMap, sync::Arc};
|
||||
use time::{OffsetDateTime, format_description};
|
||||
use tokio::io::AsyncReadExt;
|
||||
use tokio::{select, sync::mpsc};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::warn;
|
||||
@@ -41,10 +42,37 @@ use crate::client::{
|
||||
transition_api::{ReaderImpl, RequestMetadata, TransitionClient, UploadInfo},
|
||||
};
|
||||
|
||||
use crate::client::utils::base64_encode;
|
||||
use crate::client::utils::{base64_encode, base64_encode_standard};
|
||||
use rustfs_utils::path::trim_etag;
|
||||
use s3s::header::{X_AMZ_EXPIRATION, X_AMZ_VERSION_ID};
|
||||
|
||||
/// Read exactly `want` bytes for a single multipart part, or fewer if the reader
|
||||
/// reaches EOF first. Advances the reader so the next call returns the following
|
||||
/// part. Replaces the previous per-part `read_all()`/`to_vec()`, which drained
|
||||
/// the entire source into the first part and left later parts empty
|
||||
/// (rustfs/rustfs#4811).
|
||||
async fn read_multipart_part(reader: &mut ReaderImpl, want: usize) -> Result<Vec<u8>, std::io::Error> {
|
||||
match reader {
|
||||
ReaderImpl::Body(content_body) => {
|
||||
let take = content_body.len().min(want);
|
||||
Ok(content_body.split_to(take).to_vec())
|
||||
}
|
||||
ReaderImpl::ObjectBody(content_body) => {
|
||||
let mut buf = vec![0u8; want];
|
||||
let mut filled = 0;
|
||||
while filled < want {
|
||||
let n = content_body.read(&mut buf[filled..]).await?;
|
||||
if n == 0 {
|
||||
break;
|
||||
}
|
||||
filled += n;
|
||||
}
|
||||
buf.truncate(filled);
|
||||
Ok(buf)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct UploadedPartRes {
|
||||
pub error: std::io::Error,
|
||||
pub part_num: i64,
|
||||
@@ -141,14 +169,12 @@ impl TransitionClient {
|
||||
part_size = lastpart_size;
|
||||
}
|
||||
|
||||
match &mut reader {
|
||||
ReaderImpl::Body(content_body) => {
|
||||
buf = content_body.to_vec();
|
||||
}
|
||||
ReaderImpl::ObjectBody(content_body) => {
|
||||
buf = content_body.read_all().await?;
|
||||
}
|
||||
}
|
||||
// Read exactly this part's bytes. Using `read_all()`/`to_vec()` here
|
||||
// drained the whole source into the first part and left every later
|
||||
// part empty, silently corrupting any multipart upload of a streamed
|
||||
// (`ObjectBody`) source — e.g. ILM transitions of >128 MiB objects,
|
||||
// which split into 128 MiB parts (rustfs/rustfs#4811).
|
||||
buf = read_multipart_part(&mut reader, part_size as usize).await?;
|
||||
let length = buf.len();
|
||||
|
||||
if opts.send_content_md5 {
|
||||
@@ -158,14 +184,16 @@ impl TransitionClient {
|
||||
None => return Err(std::io::Error::other("MD5 hasher not initialized")),
|
||||
};
|
||||
let hash = md5_hash.hash_encode(&buf[..length]);
|
||||
md5_base64 = base64_encode(hash.as_ref());
|
||||
} else {
|
||||
// Content-MD5 must be standard base64 for the remote to accept it.
|
||||
md5_base64 = base64_encode_standard(hash.as_ref());
|
||||
} else if opts.auto_checksum.is_set() {
|
||||
let mut crc = opts.auto_checksum.hasher()?;
|
||||
crc.update(&buf[..length]);
|
||||
let csum = crc.finalize();
|
||||
|
||||
if let Ok(header_name) = HeaderName::from_bytes(opts.auto_checksum.key().as_bytes()) {
|
||||
if let Ok(header_value) = base64_encode(csum.as_ref()).parse() {
|
||||
// x-amz-checksum-* header values are standard base64 too.
|
||||
if let Ok(header_value) = base64_encode_standard(csum.as_ref()).parse() {
|
||||
custom_header.insert(header_name, header_value);
|
||||
} else {
|
||||
warn!("Failed to parse checksum value");
|
||||
@@ -174,6 +202,10 @@ impl TransitionClient {
|
||||
warn!("Invalid header name: {}", opts.auto_checksum.key());
|
||||
}
|
||||
}
|
||||
// else: neither MD5 nor a concrete additional checksum was requested,
|
||||
// so upload the part without a per-part checksum header. Guarding the
|
||||
// branch on `is_set()` avoids calling `hasher()` on `ChecksumNone`,
|
||||
// which errors with "unsupported checksum type" (rustfs/rustfs#4811).
|
||||
|
||||
let hooked = ReaderImpl::Body(Bytes::from(buf)); //newHook(BufferReader::new(buf), opts.progress);
|
||||
let mut p = UploadPartParams {
|
||||
@@ -183,7 +215,9 @@ impl TransitionClient {
|
||||
reader: hooked,
|
||||
part_number,
|
||||
md5_base64: md5_base64.clone(),
|
||||
size: part_size,
|
||||
// Use the bytes actually read, not the planned part_size, so the
|
||||
// uploaded Content-Length matches the body even on a short read.
|
||||
size: length as i64,
|
||||
//sse: opts.server_side_encryption,
|
||||
stream_sha256: !opts.disable_content_sha256,
|
||||
custom_header: custom_header.clone(),
|
||||
@@ -194,7 +228,7 @@ impl TransitionClient {
|
||||
|
||||
parts_info.entry(part_number).or_insert(obj_part);
|
||||
|
||||
total_uploaded_size += part_size as i64;
|
||||
total_uploaded_size += length as i64;
|
||||
}
|
||||
|
||||
if size > 0 && total_uploaded_size != size {
|
||||
@@ -593,9 +627,79 @@ fn collect_complete_parts(parts_info: &HashMap<i64, ObjectPart>, total_parts_cou
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{ObjectPart, collect_complete_parts};
|
||||
use super::{ObjectPart, ReaderImpl, collect_complete_parts, read_multipart_part};
|
||||
use crate::object_api::GetObjectReader;
|
||||
use bytes::Bytes;
|
||||
use std::collections::HashMap;
|
||||
|
||||
// Drive a reader through the same per-part loop the multipart stream uses and
|
||||
// collect the size of every part. Regression for rustfs/rustfs#4811: the old
|
||||
// `read_all()` per part drained the whole source into part 1.
|
||||
async fn collect_part_sizes(mut reader: ReaderImpl, total: usize, part_size: usize, last_part_size: usize) -> Vec<usize> {
|
||||
let parts = total.div_ceil(part_size);
|
||||
let mut sizes = Vec::new();
|
||||
for part_number in 1..=parts {
|
||||
let want = if part_number == parts { last_part_size } else { part_size };
|
||||
let buf = read_multipart_part(&mut reader, want).await.unwrap();
|
||||
sizes.push(buf.len());
|
||||
}
|
||||
// Nothing must remain after the planned parts are consumed.
|
||||
assert!(read_multipart_part(&mut reader, part_size).await.unwrap().is_empty());
|
||||
sizes
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn read_multipart_part_splits_streamed_object_body_evenly() {
|
||||
// 250 bytes at part_size 100 -> parts [100, 100, 50], mirroring the
|
||||
// >128 MiB / 128 MiB split from the bug report on a small deterministic
|
||||
// stream.
|
||||
let total = 250usize;
|
||||
let (mut w, r) = tokio::io::duplex(64);
|
||||
tokio::spawn(async move {
|
||||
use tokio::io::AsyncWriteExt;
|
||||
let data: Vec<u8> = (0..total).map(|i| i as u8).collect();
|
||||
w.write_all(&data).await.unwrap();
|
||||
});
|
||||
let reader = ReaderImpl::ObjectBody(GetObjectReader {
|
||||
stream: Box::new(r),
|
||||
object_info: Default::default(),
|
||||
buffered_body: None,
|
||||
body_source: Default::default(),
|
||||
});
|
||||
|
||||
let sizes = collect_part_sizes(reader, total, 100, 50).await;
|
||||
assert_eq!(sizes, vec![100, 100, 50]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn read_multipart_part_splits_in_memory_body_evenly() {
|
||||
let total = 250usize;
|
||||
let data: Vec<u8> = (0..total).map(|i| i as u8).collect();
|
||||
let reader = ReaderImpl::Body(Bytes::from(data));
|
||||
|
||||
let sizes = collect_part_sizes(reader, total, 100, 50).await;
|
||||
assert_eq!(sizes, vec![100, 100, 50]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn read_multipart_part_stops_at_eof_without_overrun() {
|
||||
// Reader shorter than the requested part size must return only what is
|
||||
// available, not block or pad.
|
||||
let (mut w, r) = tokio::io::duplex(64);
|
||||
tokio::spawn(async move {
|
||||
use tokio::io::AsyncWriteExt;
|
||||
w.write_all(&[1u8; 30]).await.unwrap();
|
||||
});
|
||||
let mut reader = ReaderImpl::ObjectBody(GetObjectReader {
|
||||
stream: Box::new(r),
|
||||
object_info: Default::default(),
|
||||
buffered_body: None,
|
||||
body_source: Default::default(),
|
||||
});
|
||||
let buf = read_multipart_part(&mut reader, 100).await.unwrap();
|
||||
assert_eq!(buf.len(), 30);
|
||||
}
|
||||
|
||||
fn parts_map(n: i64) -> HashMap<i64, ObjectPart> {
|
||||
let mut m = HashMap::new();
|
||||
for i in 1..=n {
|
||||
|
||||
Reference in New Issue
Block a user