fran/rustfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(object): extract internal put entry and ODM provenance keys (#7071)

Browse Source 
* feat(utils): add on-demand migration provenance metadata suffixes

* refactor(object): extract internal put entry from the S3 PutObject path
This commit is contained in:
Zhengchao An
2026-09-02 22:56:23 +08:00
committed by GitHub
parent 1ab6405ac9
commit 183b5c9ede
4 changed files with 1508 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/utils/src/http/metadata_compat.rs
+89
View File
@@ -88,6 +88,21 @@ pub const SUFFIX_TIER_SKIP_FV_ID: &str = "tier-skip-fvid";
/// Per-target delete-marker version ids are stored one key per target ARN.
pub const SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX: &str = "replication-delete-marker-version-";
// On-demand migration provenance. Written by the migration write-back onto
// every pulled object so operators and later tooling can tell a migrated
// object from a client write and audit where it came from. Internal keys, so
// the existing internal-key filter keeps them out of client-visible metadata.
/// Source of a pulled object as `<provider>:<bucket>`.
pub const SUFFIX_ODM_SOURCE: &str = "odm-source";
/// ETag the source reported for the pulled object, verbatim.
pub const SUFFIX_ODM_SOURCE_ETAG: &str = "odm-source-etag";
/// Source `Last-Modified` of the pulled object, RFC 3339.
pub const SUFFIX_ODM_SOURCE_LAST_MODIFIED: &str = "odm-source-last-modified";
/// Source version id of the pulled object; empty for an unversioned source.
pub const SUFFIX_ODM_SOURCE_VERSION_ID: &str = "odm-source-version-id";
/// When the object was pulled from the source, RFC 3339.
pub const SUFFIX_ODM_PULLED_AT: &str = "odm-pulled-at";
/// Case-insensitive (ASCII) check that `s` begins with `prefix`. Equivalent to
/// `s.to_lowercase().starts_with(prefix)` when `prefix` is ASCII (as both internal prefixes are),
/// but without allocating.
@@ -591,6 +606,80 @@ mod tests {
assert!(!contains_key_bytes(&meta_sys, &long_suffix));
}
const ODM_PROVENANCE_SUFFIXES: [&str; 5] = [
SUFFIX_ODM_SOURCE,
SUFFIX_ODM_SOURCE_ETAG,
SUFFIX_ODM_SOURCE_LAST_MODIFIED,
SUFFIX_ODM_SOURCE_VERSION_ID,
SUFFIX_ODM_PULLED_AT,
];
#[test]
fn odm_provenance_suffixes_are_reserved_internal_keys_under_both_prefixes() {
for suffix in ODM_PROVENANCE_SUFFIXES {
for prefix in [RUSTFS_INTERNAL_PREFIX, MINIO_INTERNAL_PREFIX] {
let key = format!("{prefix}{suffix}");
assert!(is_internal_key(&key), "{key} must be an internal key");
assert!(has_internal_suffix(&key, suffix), "{key} must match its own suffix");
assert!(has_internal_suffix(&key.to_uppercase(), suffix), "{key} must match case-insensitively");
assert_eq!(strip_internal_prefix(&key).as_deref(), Some(suffix));
}
assert!(
!is_internal_key(&format!("x-amz-meta-{suffix}")),
"{suffix} must not leak as user metadata"
);
assert!(suffix.starts_with("odm-"), "{suffix} must stay in the odm- namespace");
assert!(
suffix.bytes().all(|b| b.is_ascii_lowercase() || b == b'-'),
"{suffix} must be lowercase-hyphenated"
);
}
let distinct: std::collections::HashSet<&str> = ODM_PROVENANCE_SUFFIXES.into_iter().collect();
assert_eq!(distinct.len(), ODM_PROVENANCE_SUFFIXES.len(), "provenance suffixes must be distinct");
}
#[test]
fn odm_provenance_values_round_trip_under_both_prefixes() {
let mut metadata = HashMap::new();
insert_str(&mut metadata, SUFFIX_ODM_SOURCE, "s3:legacy-bucket".to_string());
insert_str(&mut metadata, SUFFIX_ODM_SOURCE_ETAG, "0123456789abcdef0123456789abcdef-3".to_string());
insert_str(&mut metadata, SUFFIX_ODM_SOURCE_LAST_MODIFIED, "2026-01-02T03:04:05Z".to_string());
insert_str(&mut metadata, SUFFIX_ODM_SOURCE_VERSION_ID, String::new());
insert_str(&mut metadata, SUFFIX_ODM_PULLED_AT, "2026-09-02T00:00:00Z".to_string());
assert_eq!(
metadata.len(),
2 * ODM_PROVENANCE_SUFFIXES.len(),
"every marker is written under both prefixes"
);
for suffix in ODM_PROVENANCE_SUFFIXES {
assert!(metadata.contains_key(&internal_key_rustfs(suffix)), "missing RustFS key for {suffix}");
assert!(
metadata.contains_key(&format!("{MINIO_INTERNAL_PREFIX}{suffix}")),
"missing MinIO key for {suffix}"
);
assert!(contains_key_str(&metadata, suffix));
}
assert_eq!(get_str(&metadata, SUFFIX_ODM_SOURCE).as_deref(), Some("s3:legacy-bucket"));
assert_eq!(get_str(&metadata, SUFFIX_ODM_SOURCE_VERSION_ID).as_deref(), Some(""));
assert_eq!(
get_consistent_str(&metadata, SUFFIX_ODM_SOURCE_ETAG),
Some("0123456789abcdef0123456789abcdef-3")
);
// A MinIO-only reader must still find the marker.
let minio_only: HashMap<String, String> = metadata
.iter()
.filter(|(key, _)| starts_with_ignore_ascii_case(key, MINIO_INTERNAL_PREFIX))
.map(|(key, value)| (key.clone(), value.clone()))
.collect();
assert_eq!(get_str(&minio_only, SUFFIX_ODM_PULLED_AT).as_deref(), Some("2026-09-02T00:00:00Z"));
remove_str(&mut metadata, SUFFIX_ODM_SOURCE);
assert!(!contains_key_str(&metadata, SUFFIX_ODM_SOURCE));
assert!(contains_key_str(&metadata, SUFFIX_ODM_SOURCE_ETAG));
}
#[test]
fn target_delete_marker_versions_preserve_arn_case_and_report_conflicts() {
let arn = "arn:rustfs:replication::Target:Bucket";
rustfs/src/app/object/internal_put.rs
+968
View File
@@ -0,0 +1,968 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Internal object write entry points for trusted in-process callers.
//!
//! A system write (on-demand migration write-back, future replays) must look
//! like an ordinary client write: bucket default SSE, quota, versioning,
//! Object Lock defaults, replication scheduling and creation events all apply.
//! The single-object entry point runs the same [`DefaultObjectUsecase::put_object_core`]
//! as the S3 PutObject handler; the multipart entry points mirror the S3
//! multipart handlers' policy steps against the same storage contract.
use super::*;
use crate::app::object_data_cache::invalidate_object_data_cache_after_complete_multipart_success;
use crate::app::storage_api::multipart_usecase::contract::multipart::{CompletePart, MultipartOperations as _};
use crate::app::storage_api::object_usecase::compression::is_multipart_disk_compression_enabled;
use crate::app::storage_api::object_usecase::io::WriteEncryption;
use crate::app::storage_api::object_usecase::options::{
extract_metadata_from_mime, get_complete_multipart_upload_opts_with_replication_authorization,
};
use crate::app::storage_api::object_usecase::sse::{
EncryptionKeyKind, PrepareEncryptionRequest, mark_encrypted_multipart_metadata, sse_decryption, sse_prepare_encryption,
};
use crate::capacity::record_capacity_write;
use crate::runtime_sources::NotifyInterface;
use http::HeaderName;
/// Inputs of an internal object write. Content and user metadata follow the
/// S3 request shape so the shared write path treats them exactly like a
/// client PUT: `content_headers` are the standard object headers
/// (`Content-Type`, `Cache-Control`, `Content-Encoding`, `Content-Disposition`,
/// `Content-Language`, `Expires`), `user_metadata` carries `x-amz-meta-*`
/// entries with the prefix stripped, `tags` is the `x-amz-tagging` query
/// string and `internal_metadata` holds `x-rustfs-internal-*` /
/// `x-minio-internal-*` keys written verbatim.
pub(crate) struct InternalPutContext {
pub(crate) bucket: String,
pub(crate) key: String,
/// Plaintext object length. The single-object path requires it, exactly
/// like S3 PutObject rejects an unknown `Content-Length`.
pub(crate) size: Option<u64>,
/// Lowercase hex MD5 the body must hash to; the write fails with
/// `BadDigest` otherwise and nothing is committed.
pub(crate) expected_md5_hex: Option<String>,
/// ETag to store instead of the computed one.
pub(crate) preserve_etag: Option<String>,
pub(crate) content_headers: HashMap<String, String>,
pub(crate) user_metadata: HashMap<String, String>,
pub(crate) tags: Option<String>,
pub(crate) internal_metadata: HashMap<String, String>,
/// Publish the `s3:ObjectCreated:*` event for the write.
pub(crate) emit_events: bool,
/// `userIdentity.principalId` of the creation event.
pub(crate) principal_id: &'static str,
}
const INTERNAL_PUT_METHOD_NAME: &str = "PUT";
fn api_error_from_s3(err: S3Error) -> ApiError {
ApiError {
code: err.code().clone(),
message: err.message().unwrap_or_default().to_string(),
source: None,
}
}
fn not_initialized() -> ApiError {
ApiError {
code: S3ErrorCode::InternalError,
message: "Not init".to_string(),
source: None,
}
}
/// Build the header view of an internal write so header-driven policy
/// (content-type detection, compressibility, standard metadata capture) runs
/// unchanged.
fn internal_put_headers(content_headers: &HashMap<String, String>) -> Result<HeaderMap, ApiError> {
let mut headers = HeaderMap::with_capacity(content_headers.len());
for (name, value) in content_headers {
let name = HeaderName::from_bytes(name.as_bytes())
.map_err(|err| ApiError::invalid_request(format!("invalid content header name {name:?}: {err}")))?;
let value = HeaderValue::from_str(value)
.map_err(|err| ApiError::invalid_request(format!("invalid content header value for {name}: {err}")))?;
headers.insert(name, value);
}
Ok(headers)
}
fn header_string(headers: &HeaderMap, name: &str) -> Option<String> {
headers.get(name).and_then(|value| value.to_str().ok()).map(ToOwned::to_owned)
}
fn internal_put_content_input(headers: &HeaderMap, tags: Option<String>) -> PutObjectContentInput {
PutObjectContentInput {
cache_control: header_string(headers, "cache-control"),
content_disposition: header_string(headers, "content-disposition"),
content_encoding: header_string(headers, "content-encoding"),
content_language: header_string(headers, "content-language"),
content_type: header_string(headers, "content-type"),
expires: header_string(headers, "expires"),
website_redirect_location: None,
tagging: tags,
storage_class: None,
}
}
async fn validate_internal_write_target(key: &str, bucket: &str, headers: &HeaderMap) -> Result<(), ApiError> {
validate_object_key(key, INTERNAL_PUT_METHOD_NAME).map_err(api_error_from_s3)?;
validate_table_catalog_object_mutation(bucket, key)
.await
.map_err(api_error_from_s3)?;
validate_archive_content_encoding(
key,
headers.get("content-type").and_then(|value| value.to_str().ok()),
headers.get("content-encoding").and_then(|value| value.to_str().ok()),
)
.map_err(api_error_from_s3)
}
fn internal_events_wanted() -> bool {
crate::module_switches::is_notify_module_enabled()
|| rustfs_notify::notification_system().is_some_and(|system| system.has_live_listeners())
}
/// Creation event of an internal write, published on successful completion
/// the way [`OperationHelper`] publishes it for an S3 request.
pub(super) struct InternalPutObjectEvent {
builder: EventArgsBuilder,
notify: Arc<dyn NotifyInterface>,
request_context: request_context::RequestContext,
}
impl InternalPutObjectEvent {
/// `None` when neither the notify module nor a live listener wants events.
pub(super) fn new(
notify: Arc<dyn NotifyInterface>,
request_context: request_context::RequestContext,
event_name: EventName,
bucket: &str,
key: &str,
principal_id: &'static str,
) -> Option<Self> {
if !internal_events_wanted() {
return None;
}
Some(Self {
builder: Self::builder(event_name, bucket, key, principal_id),
notify,
request_context,
})
}
fn builder(event_name: EventName, bucket: &str, key: &str, principal_id: &'static str) -> EventArgsBuilder {
// The object is a placeholder until `object()` supplies the committed
// ObjectInfo, matching the S3 helper.
let placeholder = ObjectInfo {
bucket: bucket.to_string(),
name: key.to_string(),
..Default::default()
};
EventArgsBuilder::new(event_name, bucket.to_string(), convert_ecstore_object_info(placeholder))
.req_param("principalId", principal_id)
}
pub(super) fn object(self, obj_info: ObjectInfo) -> Self {
let Self {
builder,
notify,
request_context,
} = self;
Self {
builder: builder.object(convert_ecstore_object_info(obj_info)),
notify,
request_context,
}
}
pub(super) fn version_id(self, version_id: String) -> Self {
let Self {
builder,
notify,
request_context,
} = self;
Self {
builder: builder.version_id(version_id),
notify,
request_context,
}
}
/// Publish the event when `result` is a success; failures publish nothing.
pub(super) fn complete<T>(self, result: &S3Result<S3Response<T>>) {
let Ok(response) = result else {
return;
};
let Self {
builder,
notify,
request_context,
} = self;
let event_args = builder
.resp_elements(build_event_resp_elements(response, &request_context.request_id))
.build();
spawn_background_with_context(Some(request_context), async move {
notify.notify(event_args).await;
});
}
}
impl DefaultObjectUsecase {
/// Write one object through the ordinary PutObject path on behalf of a
/// trusted internal caller.
///
/// The body is consumed exactly like a client request body: hashed
/// against `expected_md5_hex`, compressed and encrypted per bucket policy,
/// and committed under quota admission. On any failure nothing is left
/// behind.
pub(crate) async fn internal_put_object<B>(&self, ctx: InternalPutContext, body: B) -> Result<ObjectInfo, ApiError>
where
B: Stream<Item = io::Result<Bytes>> + Send + Sync + 'static,
{
let start_time = Instant::now();
let InternalPutContext {
bucket,
key,
size,
expected_md5_hex,
preserve_etag,
content_headers,
user_metadata,
tags,
internal_metadata,
emit_events,
principal_id,
} = ctx;
let Some(size) = size else {
return Err(ApiError::invalid_request("internal put requires a known object size"));
};
let size = i64::try_from(size).map_err(|_| ApiError::invalid_request("internal put size exceeds the supported range"))?;
let headers = internal_put_headers(&content_headers)?;
validate_internal_write_target(&key, &bucket, &headers).await?;
let write = PutObjectWriteRequest {
bucket,
key,
size,
quota_operation: QuotaOperation::PutObject,
ciphertext_passthrough: false,
inbound_replication_put: false,
headers: &headers,
query: None,
trailing_headers: None,
version_id: None,
sse: PutObjectSseInput {
server_side_encryption: None,
ssekms_key_id: None,
sse_customer_algorithm: None,
sse_customer_key: None,
sse_customer_key_md5: None,
},
user_metadata,
internal_metadata,
content: internal_put_content_input(&headers, tags),
object_lock: PutObjectLockInput {
legal_hold_status: None,
mode: None,
retain_until_date: None,
},
content_md5: expected_md5_hex.map(PutObjectContentMd5::Hex),
preserve_etag,
origin: PutObjectOrigin::Internal {
principal_id,
emit_events,
},
};
let committed = self
.put_object_core(write, StreamingBlob::wrap(body), start_time)
.await
.map_err(api_error_from_s3)?;
let obj_info = committed.obj_info.clone();
let result: S3Result<S3Response<()>> = Ok(S3Response::new(()));
committed.finish(&result);
Ok(obj_info)
}
/// Start a multipart upload for an internal write. The session carries the
/// same metadata, bucket default SSE session material, Object Lock
/// defaults and replication decision a client-initiated session would.
pub(crate) async fn internal_create_multipart_upload(&self, ctx: &InternalPutContext) -> Result<String, ApiError> {
let headers = internal_put_headers(&ctx.content_headers)?;
validate_internal_write_target(&ctx.key, &ctx.bucket, &headers).await?;
let store = self.object_store().ok_or_else(not_initialized)?;
let mut metadata = ctx.user_metadata.clone();
namespace_reserved_user_metadata(&mut metadata);
extract_metadata_from_mime(&headers, &mut metadata);
if let Some(tags) = ctx.tags.clone() {
metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags);
}
let object_lock_config_state = load_bucket_object_lock_config_state(&ctx.bucket)
.await
.map_err(api_error_from_s3)?;
apply_bucket_default_lock_retention(&ctx.bucket, &object_lock_config_state, &mut metadata, false)
.map_err(api_error_from_s3)?;
// Internal callers carry no credential; a bucket default of SSE-KMS is
// authorized as an internal write, like every other system write.
let prepared_material = sse_prepare_encryption(PrepareEncryptionRequest {
bucket: &ctx.bucket,
key: &ctx.key,
server_side_encryption: None,
ssekms_key_id: None,
ssekms_context: None,
sse_customer_algorithm: None,
sse_customer_key: None,
sse_customer_key_md5: None,
principal: None,
})
.await?;
if let Some(material) = prepared_material {
let mut encryption_metadata = encryption_material_to_metadata(&material)?;
if material.key_kind == EncryptionKeyKind::Object {
mark_encrypted_multipart_metadata(&mut encryption_metadata);
}
metadata.extend(encryption_metadata);
}
if is_multipart_disk_compression_enabled() && is_disk_compressible(&headers, &ctx.key) {
insert_str(
&mut metadata,
SUFFIX_COMPRESSION,
compression_metadata_value(CompressionAlgorithm::default()),
);
}
metadata.extend(ctx.internal_metadata.clone());
let mt2 = metadata.clone();
let mut opts = put_opts_with_replication_authorization(&ctx.bucket, &ctx.key, None, &headers, metadata, false)
.await
.map_err(ApiError::from)?;
let dsc = must_replicate_object(
&ctx.bucket,
&ctx.key,
&mt2,
"".to_string(),
opts.delete_marker_replication_status(),
opts.clone(),
)
.await;
if dsc.replicate_any() {
insert_str(&mut opts.user_defined, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string());
insert_str(
&mut opts.user_defined,
SUFFIX_REPLICATION_STATUS,
dsc.pending_status().unwrap_or_default(),
);
}
let current_opts = get_opts(&ctx.bucket, &ctx.key, opts.version_id.clone(), None, &headers)
.await
.map_err(ApiError::from)?;
match store.get_object_info(&ctx.bucket, &ctx.key, &current_opts).await {
Ok(existing_obj_info) => {
validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &opts)
.map_err(api_error_from_s3)?;
}
Err(err) => {
if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
return Err(ApiError::from(err));
}
}
}
let upload = store
.new_multipart_upload(&ctx.bucket, &ctx.key, &opts)
.await
.map_err(ApiError::from)?;
Ok(upload.upload_id)
}
/// Stage one part of an internal multipart upload. Compression and managed
/// SSE follow the session metadata recorded at creation; the staged part
/// is verified against `expected_md5_hex` when given.
pub(crate) async fn internal_upload_part<B>(
&self,
ctx: &InternalPutContext,
upload_id: &str,
part_number: usize,
size: u64,
expected_md5_hex: Option<String>,
body: B,
) -> Result<CompletePart, ApiError>
where
B: Stream<Item = io::Result<Bytes>> + Send + Sync + Unpin + 'static,
{
let size =
i64::try_from(size).map_err(|_| ApiError::invalid_request("internal part size exceeds the supported range"))?;
let bucket = ctx.bucket.as_str();
let key = ctx.key.as_str();
let store = self.object_store().ok_or_else(not_initialized)?;
let mut opts = ObjectOptions::default();
let session = store
.get_multipart_info(bucket, key, upload_id, &opts)
.await
.map_err(ApiError::from)?;
let upload_part_admission = match get_concurrency_manager()
.admit_multipart_part(size)
.await
.map_err(|_| ApiError::other(io::Error::other("foreground write admission closed")))?
{
ForegroundWriteAdmission::Disabled => None,
ForegroundWriteAdmission::Admitted(permit) => {
counter!("rustfs.upload_part.foreground_admission.total", "result" => "admitted").increment(1);
Some(permit)
}
ForegroundWriteAdmission::Rejected => {
counter!("rustfs.upload_part.foreground_admission.total", "result" => "rejected").increment(1);
return Err(ApiError {
code: S3ErrorCode::SlowDown,
message: "foreground write concurrency limit reached, please reduce your request rate".to_string(),
source: None,
});
}
};
let buffer_size = get_buffer_size_opt_in(size);
let body = tokio::io::BufReader::with_capacity(buffer_size, StreamReader::new(body));
let actual_size = size;
let mut write_plan = WritePlan::new();
let mut reader = if rustfs_utils::http::contains_key_str(&session.user_defined, SUFFIX_COMPRESSION) {
let hrd = HashReader::from_stream(body, size, actual_size, expected_md5_hex, None, false).map_err(ApiError::from)?;
write_plan = write_plan.with_compression(CompressionAlgorithm::default());
hrd
} else {
HashReader::from_stream(body, size, actual_size, expected_md5_hex, None, false).map_err(ApiError::from)?
};
opts.want_checksum = reader.checksum();
if session
.user_defined
.contains_key("x-amz-server-side-encryption-customer-algorithm")
{
return Err(ApiError::invalid_request("internal multipart writes cannot continue an SSE-C session"));
}
if session.user_defined.contains_key("x-amz-server-side-encryption") {
// Reuses the envelope prepared at creation; the session pins the key.
let managed_material = sse_decryption(DecryptionRequest {
bucket,
key,
metadata: &session.user_defined,
sse_customer_key: None,
sse_customer_key_md5: None,
principal: None,
})
.await?
.ok_or_else(|| ApiError::from(StorageError::other("Missing managed SSE session material")))?;
let managed_write = match managed_material.key_kind {
EncryptionKeyKind::Object => {
WriteEncryption::multipart_object_key(managed_material.key_bytes, part_number as u32)
}
EncryptionKeyKind::Direct => {
WriteEncryption::multipart(managed_material.key_bytes, managed_material.base_nonce, part_number)
}
};
write_plan = write_plan.with_encryption(managed_write);
}
reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?;
let mut reader = PutObjReader::new(reader);
let _upload_part_admission = upload_part_admission;
let info = store
.put_object_part(bucket, key, upload_id, part_number, &mut reader, &opts)
.await
.map_err(ApiError::from)?;
drop(_upload_part_admission);
Ok(CompletePart {
part_num: info.part_num,
etag: info.etag,
..Default::default()
})
}
/// Complete an internal multipart upload: versioning, quota admission,
/// Object Lock validation of the overwritten version, usage accounting,
/// immediate ILM transition, replication scheduling and the creation
/// event, as the S3 CompleteMultipartUpload handler performs them.
pub(crate) async fn internal_complete_multipart_upload(
&self,
ctx: &InternalPutContext,
upload_id: &str,
parts: Vec<CompletePart>,
) -> Result<ObjectInfo, ApiError> {
let bucket = ctx.bucket.clone();
let key = ctx.key.clone();
if parts.is_empty() {
return Err(ApiError::invalid_request("You must specify at least one part"));
}
if parts.windows(2).any(|pair| pair[0].part_num >= pair[1].part_num) {
return Err(ApiError::invalid_request("multipart parts must be listed in ascending part number order"));
}
validate_table_catalog_object_mutation(&bucket, &key)
.await
.map_err(api_error_from_s3)?;
let store = self.object_store().ok_or_else(not_initialized)?;
let headers = HeaderMap::new();
let mut opts =
get_complete_multipart_upload_opts_with_replication_authorization(&headers, false).map_err(ApiError::from)?;
opts.preserve_etag = ctx.preserve_etag.clone();
let versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
opts.versioned = versioned;
opts.version_suspended = BucketVersioningSys::prefix_suspended(&bucket, &key).await;
let capacity_scope_token = Uuid::new_v4();
opts.capacity_scope_token = Some(capacity_scope_token);
let current_opts =
internal_object_info_lookup_opts(get_opts(&bucket, &key, None, None, &headers).await.map_err(ApiError::from)?);
let object_lock_config_state = load_bucket_object_lock_config_state(&bucket)
.await
.map_err(api_error_from_s3)?;
let previous_current_sizes = match store.get_object_info(&bucket, &key, &current_opts).await {
Ok(existing_obj_info) => {
validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &current_opts)
.map_err(api_error_from_s3)?;
let physical_size = existing_obj_info.size.max(0) as u64;
let logical_size = quota_object_size(&existing_obj_info);
Some((physical_size, logical_size))
}
Err(err) => {
if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
return Err(ApiError::from(err));
}
None
}
};
let cache_adapter = self.object_data_cache();
let _ = invalidate_object_data_cache_before_mutation(&cache_adapter, &bucket, &key).await;
let quota_metadata_sys = self.bucket_metadata_sys();
let quota_tracking = quota_metadata_sys.is_some();
let mut quota_enabled = false;
if let Some(metadata_sys) = quota_metadata_sys {
let quota_checker = QuotaChecker::new(metadata_sys);
let check_result =
map_quota_check_outcome(&bucket, quota_checker.check_quota(&bucket, QuotaOperation::PutObject, 0).await)
.map_err(api_error_from_s3)?;
quota_enabled = check_result.quota_limit.is_some();
apply_quota_admission(&mut opts, &check_result).map_err(api_error_from_s3)?;
}
let previous_current_size = match previous_current_sizes {
Some((_, Ok(logical_size))) if quota_enabled => Some(logical_size),
Some((_, Err(err))) if quota_enabled => return Err(ApiError::from(err)),
Some((physical_size, _)) => Some(physical_size),
None => None,
};
let event = ctx.emit_events.then(|| {
InternalPutObjectEvent::new(
current_notify_interface_for_context(self.context.as_deref()),
request_context::RequestContext::fallback(),
EventName::ObjectCreatedCompleteMultipartUpload,
&bucket,
&key,
ctx.principal_id,
)
});
// The spawned task owns the commit so a cancelled caller cannot leave
// the bookkeeping half done.
let complete_commit = spawn_traced_join({
let store = Arc::clone(&store);
let bucket = bucket.clone();
let key = key.clone();
let upload_id = upload_id.to_string();
let opts = opts.clone();
async move {
let obj_info = store
.clone()
.complete_multipart_upload(&bucket, &key, &upload_id, parts, &opts)
.await
.map_err(ApiError::from)?;
let _ = invalidate_object_data_cache_after_complete_multipart_success(&cache_adapter, &bucket, &key).await;
record_capacity_write(Some(capacity_scope_token)).await;
if quota_tracking {
let committed_size = quota_accounting_object_size(&obj_info, quota_enabled).map_err(api_error_from_s3)?;
if versioned {
record_bucket_object_version_write_memory(&bucket, previous_current_size, committed_size).await;
} else {
record_bucket_object_write_memory(&bucket, previous_current_size, committed_size).await;
}
}
enqueue_transition_immediate(&obj_info, LcEventSrc::S3CompleteMultipartUpload).await;
let mt2 = obj_info.user_defined.clone();
let dsc = must_replicate_object(
&bucket,
&key,
&mt2,
"".to_string(),
opts.delete_marker_replication_status(),
opts.clone(),
)
.await;
if dsc.replicate_any() {
schedule_object_replication(obj_info.clone(), store, dsc).await;
}
rustfs_scanner::record_dirty_usage_bucket(&bucket);
Ok::<_, ApiError>(obj_info)
}
});
let obj_info = complete_commit.await.map_err(|err| {
ApiError::other(io::Error::other(format!("complete multipart upload commit owner task failed: {err}")))
})??;
if let Some(event) = event.flatten() {
let mut event = event.object(obj_info.clone());
if versioned && let Some(version_id) = obj_info.version_id {
event = event.version_id(version_id.to_string());
}
let result: S3Result<S3Response<()>> = Ok(S3Response::new(()));
event.complete(&result);
}
Ok(obj_info)
}
/// Discard an internal multipart upload and its staged parts.
pub(crate) async fn internal_abort_multipart_upload(&self, bucket: &str, key: &str, upload_id: &str) -> Result<(), ApiError> {
let store = self.object_store().ok_or_else(not_initialized)?;
store
.abort_multipart_upload(bucket, key, upload_id, &ObjectOptions::default())
.await
.map_err(ApiError::from)?;
rustfs_scanner::record_dirty_usage_bucket(bucket);
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
use rustfs_utils::http::{
MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX, SUFFIX_ODM_PULLED_AT, SUFFIX_ODM_SOURCE, SUFFIX_ODM_SOURCE_ETAG,
SUFFIX_ODM_SOURCE_LAST_MODIFIED, SUFFIX_ODM_SOURCE_VERSION_ID, contains_key_str, get_str,
};
const TEST_PRINCIPAL: &str = "rustfs-internal-put-test";
fn md5_hex(body: &[u8]) -> String {
hex_simd::encode_to_string(Md5::digest(body), hex_simd::AsciiCase::Lower)
}
fn body_stream(chunks: Vec<Bytes>) -> impl Stream<Item = io::Result<Bytes>> + Send + Sync + Unpin + 'static {
futures::stream::iter(chunks.into_iter().map(Ok))
}
fn provenance_metadata() -> HashMap<String, String> {
let mut internal_metadata = HashMap::new();
insert_str(&mut internal_metadata, SUFFIX_ODM_SOURCE, "s3:source-bucket".to_string());
insert_str(
&mut internal_metadata,
SUFFIX_ODM_SOURCE_ETAG,
"\"0123456789abcdef0123456789abcdef-3\"".to_string(),
);
insert_str(
&mut internal_metadata,
SUFFIX_ODM_SOURCE_LAST_MODIFIED,
"2026-01-02T03:04:05Z".to_string(),
);
insert_str(&mut internal_metadata, SUFFIX_ODM_SOURCE_VERSION_ID, String::new());
insert_str(&mut internal_metadata, SUFFIX_ODM_PULLED_AT, "2026-09-02T00:00:00Z".to_string());
internal_metadata
}
fn internal_context(bucket: &str, key: &str, body: &[u8]) -> InternalPutContext {
InternalPutContext {
bucket: bucket.to_string(),
key: key.to_string(),
size: Some(body.len() as u64),
expected_md5_hex: Some(md5_hex(body)),
preserve_etag: None,
content_headers: HashMap::from([
("Content-Type".to_string(), "text/plain".to_string()),
("Cache-Control".to_string(), "max-age=60".to_string()),
]),
user_metadata: HashMap::from([("origin".to_string(), "unit-test".to_string())]),
tags: Some("team=storage".to_string()),
internal_metadata: provenance_metadata(),
emit_events: false,
principal_id: TEST_PRINCIPAL,
}
}
async fn internal_put_test_bucket(prefix: &str) -> (Arc<ECStore>, String) {
let store = crate::app::gating_test_env::shared_gating_ecstore().await;
crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await;
let bucket = format!("{prefix}-{}", Uuid::new_v4().simple());
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create internal put test bucket");
(store, bucket)
}
#[tokio::test]
#[serial_test::serial]
async fn internal_put_object_writes_through_the_shared_put_path() {
let (store, bucket) = internal_put_test_bucket("internal-put").await;
let body = b"internal write-back body".to_vec();
let ctx = internal_context(&bucket, "dir/object.txt", &body);
let obj_info = DefaultObjectUsecase::from_global()
.internal_put_object(ctx, body_stream(vec![Bytes::from(body.clone())]))
.await
.expect("internal put must succeed");
assert_eq!(obj_info.etag.as_deref(), Some(md5_hex(&body).as_str()));
assert_eq!(obj_info.size, body.len() as i64);
let stored = store
.get_object_info(&bucket, "dir/object.txt", &ObjectOptions::default())
.await
.expect("internal put must leave a readable object");
let metadata = &stored.user_defined;
assert_eq!(metadata.get("content-type").map(String::as_str), Some("text/plain"));
assert_eq!(metadata.get("cache-control").map(String::as_str), Some("max-age=60"));
assert_eq!(metadata.get("origin").map(String::as_str), Some("unit-test"));
assert_eq!(stored.user_tags.as_str(), "team=storage", "tags must be committed as object tags");
for suffix in [
SUFFIX_ODM_SOURCE,
SUFFIX_ODM_SOURCE_ETAG,
SUFFIX_ODM_SOURCE_LAST_MODIFIED,
SUFFIX_ODM_SOURCE_VERSION_ID,
SUFFIX_ODM_PULLED_AT,
] {
assert!(
metadata.contains_key(&format!("{RUSTFS_INTERNAL_PREFIX}{suffix}")),
"missing RustFS provenance key {suffix}"
);
assert!(
metadata.contains_key(&format!("{MINIO_INTERNAL_PREFIX}{suffix}")),
"missing MinIO provenance key {suffix}"
);
}
assert_eq!(get_str(metadata, SUFFIX_ODM_SOURCE).as_deref(), Some("s3:source-bucket"));
}
#[tokio::test]
#[serial_test::serial]
async fn internal_put_object_preserves_the_caller_etag() {
let (store, bucket) = internal_put_test_bucket("internal-put-etag").await;
let body = b"etag is preserved verbatim".to_vec();
let mut ctx = internal_context(&bucket, "preserved.bin", &body);
ctx.preserve_etag = Some("0123456789abcdef0123456789abcdef-3".to_string());
let obj_info = DefaultObjectUsecase::from_global()
.internal_put_object(ctx, body_stream(vec![Bytes::from(body.clone())]))
.await
.expect("internal put with a preserved ETag must succeed");
assert_eq!(obj_info.etag.as_deref(), Some("0123456789abcdef0123456789abcdef-3"));
let stored = store
.get_object_info(&bucket, "preserved.bin", &ObjectOptions::default())
.await
.expect("preserved-ETag object must be readable");
assert_eq!(stored.etag.as_deref(), Some("0123456789abcdef0123456789abcdef-3"));
}
#[tokio::test]
#[serial_test::serial]
async fn internal_put_object_rejects_a_digest_mismatch_without_committing() {
let (store, bucket) = internal_put_test_bucket("internal-put-digest").await;
let body = b"body whose digest will not match".to_vec();
let mut ctx = internal_context(&bucket, "mismatch.bin", &body);
ctx.expected_md5_hex = Some(md5_hex(b"a different body"));
let err = DefaultObjectUsecase::from_global()
.internal_put_object(ctx, body_stream(vec![Bytes::from(body)]))
.await
.expect_err("digest mismatch must fail the internal put");
assert_eq!(err.code, S3ErrorCode::BadDigest, "unexpected error: {err}");
let lookup = store
.get_object_info(&bucket, "mismatch.bin", &ObjectOptions::default())
.await;
assert!(
lookup.as_ref().is_err_and(is_err_object_not_found),
"a rejected internal put must not leave an object behind"
);
}
#[tokio::test]
#[serial_test::serial]
async fn internal_put_object_requires_a_known_size() {
let (_store, bucket) = internal_put_test_bucket("internal-put-size").await;
let body = b"unknown length".to_vec();
let mut ctx = internal_context(&bucket, "unknown.bin", &body);
ctx.size = None;
let err = DefaultObjectUsecase::from_global()
.internal_put_object(ctx, body_stream(vec![Bytes::from(body)]))
.await
.expect_err("an unknown size must be rejected before the body is read");
assert_eq!(err.code, S3ErrorCode::InvalidRequest);
}
#[tokio::test]
#[serial_test::serial]
async fn internal_multipart_roundtrip_completes_and_abort_leaves_nothing() {
const FIRST_PART_SIZE: usize = 5 * 1024 * 1024;
let (store, bucket) = internal_put_test_bucket("internal-mpu").await;
let usecase = DefaultObjectUsecase::from_global();
let first_part = vec![0x41u8; FIRST_PART_SIZE];
let last_part = b"tail of the multipart object".to_vec();
let mut ctx = internal_context(&bucket, "multipart/object.bin", &[]);
ctx.size = None;
ctx.expected_md5_hex = None;
ctx.preserve_etag = Some("0123456789abcdef0123456789abcdef-2".to_string());
let upload_id = usecase
.internal_create_multipart_upload(&ctx)
.await
.expect("internal multipart create must succeed");
let part_one = usecase
.internal_upload_part(
&ctx,
&upload_id,
1,
first_part.len() as u64,
Some(md5_hex(&first_part)),
body_stream(vec![Bytes::from(first_part.clone())]),
)
.await
.expect("first internal part must stage");
let part_two = usecase
.internal_upload_part(
&ctx,
&upload_id,
2,
last_part.len() as u64,
Some(md5_hex(&last_part)),
body_stream(vec![Bytes::from(last_part.clone())]),
)
.await
.expect("last internal part must stage");
assert_eq!(part_one.part_num, 1);
assert_eq!(part_two.part_num, 2);
let obj_info = usecase
.internal_complete_multipart_upload(&ctx, &upload_id, vec![part_one, part_two])
.await
.expect("internal multipart complete must succeed");
assert_eq!(obj_info.size, (first_part.len() + last_part.len()) as i64);
assert_eq!(obj_info.parts.len(), 2);
assert_eq!(obj_info.etag.as_deref(), Some("0123456789abcdef0123456789abcdef-2"));
let stored = store
.get_object_info(&bucket, &ctx.key, &ObjectOptions::default())
.await
.expect("completed multipart object must be readable");
assert_eq!(stored.user_defined.get("content-type").map(String::as_str), Some("text/plain"));
assert_eq!(stored.user_defined.get("origin").map(String::as_str), Some("unit-test"));
assert!(contains_key_str(&stored.user_defined, SUFFIX_ODM_SOURCE));
let aborted_upload_id = usecase
.internal_create_multipart_upload(&ctx)
.await
.expect("second internal multipart create must succeed");
usecase
.internal_upload_part(
&ctx,
&aborted_upload_id,
1,
last_part.len() as u64,
None,
body_stream(vec![Bytes::from(last_part.clone())]),
)
.await
.expect("part of the aborted upload must stage");
usecase
.internal_abort_multipart_upload(&bucket, &ctx.key, &aborted_upload_id)
.await
.expect("internal abort must succeed");
let uploads = store
.list_multipart_uploads(&bucket, &ctx.key, None, None, None, 100)
.await
.expect("list multipart uploads after abort");
assert!(
uploads.uploads.iter().all(|upload| upload.upload_id != aborted_upload_id),
"aborted internal upload must not linger: {:?}",
uploads.uploads
);
}
#[tokio::test]
#[serial_test::serial]
async fn internal_complete_multipart_upload_rejects_unordered_parts() {
let (_store, bucket) = internal_put_test_bucket("internal-mpu-order").await;
let ctx = internal_context(&bucket, "unordered.bin", &[]);
let parts = vec![
CompletePart {
part_num: 2,
..Default::default()
},
CompletePart {
part_num: 1,
..Default::default()
},
];
let err = DefaultObjectUsecase::from_global()
.internal_complete_multipart_upload(&ctx, "upload", parts)
.await
.expect_err("unordered parts must be rejected before touching the store");
assert_eq!(err.code, S3ErrorCode::InvalidRequest);
}
#[test]
fn internal_put_event_names_the_principal_and_creation_event() {
let event_args = InternalPutObjectEvent::builder(EventName::ObjectCreatedPut, "bucket", "key", TEST_PRINCIPAL).build();
assert_eq!(event_args.event_name, EventName::ObjectCreatedPut);
assert_eq!(event_args.bucket_name, "bucket");
assert_eq!(event_args.req_params.get("principalId").map(String::as_str), Some(TEST_PRINCIPAL));
assert!(!event_args.is_replication_request());
}
#[test]
fn internal_put_headers_normalize_names_and_reject_invalid_values() {
let headers = internal_put_headers(&HashMap::from([
("Content-Type".to_string(), "application/json".to_string()),
("Cache-Control".to_string(), "no-cache".to_string()),
]))
.expect("valid content headers must build");
assert_eq!(headers.get("content-type").and_then(|v| v.to_str().ok()), Some("application/json"));
let content = internal_put_content_input(&headers, Some("a=b".to_string()));
assert_eq!(content.content_type.as_deref(), Some("application/json"));
assert_eq!(content.cache_control.as_deref(), Some("no-cache"));
assert_eq!(content.tagging.as_deref(), Some("a=b"));
assert!(content.storage_class.is_none());
let err = internal_put_headers(&HashMap::from([("Content-Type".to_string(), "bad\nvalue".to_string())]))
.expect_err("a header value with a control character must be rejected");
assert_eq!(err.code, S3ErrorCode::InvalidRequest);
}
}
rustfs/src/app/object/mod.rs
+5
View File
@@ -191,6 +191,10 @@ mod delete;
mod extract;
mod get;
mod head;
// Consumed by the on-demand migration write-back (rustfs/backlog#2153); until
// that lands only tests construct the internal entry points.
#[cfg_attr(not(test), expect(dead_code, reason = "wired by the on-demand migration write-back"))]
mod internal_put;
mod put;
mod restore;
mod shared;
@@ -202,6 +206,7 @@ pub(crate) use self::copy::*;
pub(crate) use self::delete::*;
pub(crate) use self::extract::*;
pub(crate) use self::get::*;
pub(crate) use self::internal_put::*;
use self::put::*;
pub(crate) use self::put::{guard_put_object_body_read_timeout, put_object_body_read_timeout};
pub(crate) use self::shared::*;
rustfs/src/app/object/put.rs
+441 -122
View File
@@ -895,6 +895,226 @@ fn is_post_object_sse_kms_requested(input: &PutObjectInput, headers: &HeaderMap)
is_sse_kms_requested(input, headers)
}
/// Standard content headers and the tagging / storage-class values of a PUT
/// that become object metadata through [`apply_put_request_metadata`].
pub(super) struct PutObjectContentInput {
pub(super) cache_control: Option<CacheControl>,
pub(super) content_disposition: Option<ContentDisposition>,
pub(super) content_encoding: Option<ContentEncoding>,
pub(super) content_language: Option<ContentLanguage>,
pub(super) content_type: Option<ContentType>,
pub(super) expires: Option<String>,
pub(super) website_redirect_location: Option<WebsiteRedirectLocation>,
pub(super) tagging: Option<TaggingHeader>,
pub(super) storage_class: Option<StorageClass>,
}
/// Encryption inputs of a PUT after the S3 input/header merge.
pub(super) struct PutObjectSseInput {
pub(super) server_side_encryption: Option<ServerSideEncryption>,
pub(super) ssekms_key_id: Option<SSEKMSKeyId>,
pub(super) sse_customer_algorithm: Option<SSECustomerAlgorithm>,
pub(super) sse_customer_key: Option<s3s::dto::SSECustomerKey>,
pub(super) sse_customer_key_md5: Option<SSECustomerKeyMD5>,
}
/// Explicit Object Lock values of a PUT; all `None` means the bucket default
/// retention decides.
pub(super) struct PutObjectLockInput {
pub(super) legal_hold_status: Option<ObjectLockLegalHoldStatus>,
pub(super) mode: Option<ObjectLockMode>,
pub(super) retain_until_date: Option<Timestamp>,
}
/// Expected body MD5 as the caller carries it. It is decoded at the same
/// point of the write path for every origin so an invalid digest keeps its
/// precedence relative to quota, admission and Object Lock errors.
pub(super) enum PutObjectContentMd5 {
/// `Content-MD5` request header value.
Base64(String),
/// Lowercase hex digest, as an internal caller already holds it.
#[cfg_attr(not(test), expect(dead_code, reason = "constructed by the internal put entry point"))]
Hex(String),
}
/// Where a single-object write originates.
pub(super) enum PutObjectOrigin<'a> {
/// The S3 PutObject/PostObject handler: request-bound identity, the
/// audit/notification chain and the bucket-generation guard installed by
/// the access layer all come from the request.
S3 {
req: &'a S3Request<PutObjectInput>,
event_name: EventName,
},
/// A trusted in-process caller writing on the server's behalf. There is no
/// request and no credential: managed-SSE authorization treats the write
/// as internal, and the creation event, when requested, names
/// `principal_id` instead of an access key.
#[cfg_attr(not(test), expect(dead_code, reason = "constructed by the internal put entry point"))]
Internal { principal_id: &'static str, emit_events: bool },
}
impl PutObjectOrigin<'_> {
fn replication_request_authorized(&self) -> bool {
match self {
Self::S3 { req, .. } => replication_request_authorized(req),
Self::Internal { .. } => false,
}
}
fn apply_bucket_generation_guard(&self, bucket: &str, opts: &mut ObjectOptions) -> S3Result<()> {
match self {
Self::S3 { req, .. } => apply_bucket_generation_guard(req, bucket, opts),
Self::Internal { .. } => Ok(()),
}
}
fn sse_principal(&self) -> Option<SseKmsPrincipal> {
match self {
Self::S3 { req, .. } => SseKmsPrincipal::from_request(req),
Self::Internal { .. } => None,
}
}
}
/// Every input the shared single-object write path needs, independent of
/// whether an S3 request or an internal caller produced it.
pub(super) struct PutObjectWriteRequest<'a> {
pub(super) bucket: String,
pub(super) key: String,
/// Authoritative plaintext length; never negative.
pub(super) size: i64,
pub(super) quota_operation: QuotaOperation,
/// Authorized SSE-C replication body that is already ciphertext.
pub(super) ciphertext_passthrough: bool,
pub(super) inbound_replication_put: bool,
/// Request headers, or the object's content headers for an internal write.
pub(super) headers: &'a HeaderMap,
pub(super) query: Option<&'a str>,
pub(super) trailing_headers: Option<s3s::TrailingHeaders>,
pub(super) version_id: Option<String>,
pub(super) sse: PutObjectSseInput,
/// User metadata keyed the way s3s delivers it (`x-amz-meta-` stripped).
pub(super) user_metadata: HashMap<String, String>,
/// Internal `x-rustfs-internal-*` / `x-minio-internal-*` keys written
/// verbatim onto the object; empty for S3 requests.
pub(super) internal_metadata: HashMap<String, String>,
pub(super) content: PutObjectContentInput,
pub(super) object_lock: PutObjectLockInput,
pub(super) content_md5: Option<PutObjectContentMd5>,
/// ETag to store instead of the computed one; `None` keeps the computed
/// (or replication-header-derived) value.
pub(super) preserve_etag: Option<String>,
pub(super) origin: PutObjectOrigin<'a>,
}
/// Audit/notification completion of a write, per origin.
pub(super) enum PutObjectCompletion {
S3(OperationHelper),
Internal(Option<Box<InternalPutObjectEvent>>),
}
impl PutObjectCompletion {
fn object(self, obj_info: ObjectInfo) -> Self {
match self {
Self::S3(helper) => Self::S3(helper.object(obj_info)),
Self::Internal(event) => Self::Internal(event.map(|event| Box::new(event.object(obj_info)))),
}
}
fn version_id(self, version_id: String) -> Self {
match self {
Self::S3(helper) => Self::S3(helper.version_id(version_id)),
Self::Internal(event) => Self::Internal(event.map(|event| Box::new(event.version_id(version_id)))),
}
}
fn complete<T>(self, result: &S3Result<S3Response<T>>) -> Self {
match self {
Self::S3(helper) => Self::S3(helper.complete(result)),
Self::Internal(event) => {
if let Some(event) = event {
event.complete(result);
}
Self::Internal(None)
}
}
}
}
/// Record the failed completion and hand the error back to the caller.
fn fail_put_object(completion: PutObjectCompletion, err: S3Error) -> S3Error {
let result: S3Result<S3Response<()>> = Err(err);
let _ = completion.complete(&result);
match result {
Err(err) => err,
Ok(_) => unreachable!("failed PutObject completion carries an error"),
}
}
/// A committed single-object write awaiting its response-side completion.
pub(super) struct PutObjectCommitted {
pub(super) obj_info: ObjectInfo,
pub(super) put_versioned: bool,
pub(super) effective_sse: Option<ServerSideEncryption>,
pub(super) effective_kms_key_id: Option<SSEKMSKeyId>,
pub(super) sse_customer_algorithm: Option<SSECustomerAlgorithm>,
pub(super) sse_customer_key_md5: Option<SSECustomerKeyMD5>,
pub(super) put_extra_checksum_headers: Vec<(&'static str, String)>,
completion: PutObjectCompletion,
put_request_guard: PutObjectGuard,
bucket: String,
key: String,
start_time: Instant,
size: i64,
use_zero_copy_eager_put_path: bool,
concurrent_put_requests: usize,
buffer_size: usize,
}
impl PutObjectCommitted {
/// Publish the audit entry / creation event for `result`, record the
/// request-level PutObject metrics and release the request guard.
pub(super) fn finish<T>(self, result: &S3Result<S3Response<T>>) {
let Self {
completion,
mut put_request_guard,
bucket,
key,
start_time,
size,
use_zero_copy_eager_put_path,
concurrent_put_requests,
buffer_size,
..
} = self;
let _ = completion.complete(result);
// Record PutObject metrics via zero-copy-metrics
{
let duration_ms = start_time.elapsed().as_millis() as f64;
rustfs_io_metrics::record_put_object(
duration_ms,
size,
use_zero_copy_eager_put_path, // Track if zero-copy was enabled
);
}
debug!(
target: "rustfs::app::object_usecase",
component = "app",
subsystem = "object",
bucket = %bucket,
key = %key,
concurrent_put_requests,
buffer_size,
"PutObject request completed"
);
put_request_guard.finish_ok();
}
}
impl DefaultObjectUsecase {
fn should_use_large_put_concurrency_tuning(size: i64) -> bool {
size >= DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES
@@ -1055,7 +1275,7 @@ impl DefaultObjectUsecase {
};
// Resolve the authoritative decoded/plain object length (rejecting negative/unknown) before anything else consumes it.
let mut size = resolve_put_object_authoritative_size(&req.headers, content_length)?;
let size = resolve_put_object_authoritative_size(&req.headers, content_length)?;
if let Some(limit) = max_content_length
&& u64::try_from(size).is_ok_and(|size| size > limit)
@@ -1063,6 +1283,140 @@ impl DefaultObjectUsecase {
return Err(S3Error::new(S3ErrorCode::EntityTooLarge));
}
let write = PutObjectWriteRequest {
bucket: bucket.clone(),
key,
size,
quota_operation,
ciphertext_passthrough,
inbound_replication_put,
headers: &req.headers,
query: req.uri.query(),
trailing_headers: req.trailing_headers.clone(),
version_id,
sse: PutObjectSseInput {
server_side_encryption,
ssekms_key_id,
sse_customer_algorithm,
sse_customer_key,
sse_customer_key_md5,
},
user_metadata: metadata.unwrap_or_default(),
internal_metadata: HashMap::new(),
content: PutObjectContentInput {
cache_control,
content_disposition,
content_encoding,
content_language,
content_type,
expires,
website_redirect_location,
tagging,
storage_class,
},
object_lock: PutObjectLockInput {
legal_hold_status: object_lock_legal_hold_status,
mode: object_lock_mode,
retain_until_date: object_lock_retain_until_date,
},
content_md5: content_md5.map(PutObjectContentMd5::Base64),
preserve_etag: None,
origin: PutObjectOrigin::S3 { req: &req, event_name },
};
let committed = self.put_object_core(write, body, start_time).await?;
let raw_version = committed.obj_info.version_id.map(|v| v.to_string());
let put_version = if committed.put_versioned { raw_version } else { None };
let e_tag = committed.obj_info.etag.clone().map(|etag| to_s3s_etag(&etag));
let expiration = resolve_put_object_expiration(&bucket, &committed.obj_info).await;
let mut checksums = PutObjectChecksums {
crc32: input.checksum_crc32,
crc32c: input.checksum_crc32c,
sha1: input.checksum_sha1,
sha256: input.checksum_sha256,
crc64nvme: input.checksum_crc64nvme,
};
apply_trailing_checksums(
input.checksum_algorithm.as_ref().map(|a| a.as_str()),
&req.trailing_headers,
&mut checksums,
);
let output = PutObjectOutput {
e_tag,
server_side_encryption: committed.effective_sse.clone(),
sse_customer_algorithm: committed.sse_customer_algorithm.clone(),
sse_customer_key_md5: committed.sse_customer_key_md5.clone(),
ssekms_key_id: committed.effective_kms_key_id.clone(),
expiration,
checksum_crc32: checksums.crc32,
checksum_crc32c: checksums.crc32c,
checksum_sha1: checksums.sha1,
checksum_sha256: checksums.sha256,
checksum_crc64nvme: checksums.crc64nvme,
version_id: put_version,
..Default::default()
};
// For browser-based POST uploads (multipart/form-data), response status/body handling
// is decided by s3s PostObject serializer (success_action_status / redirect semantics).
let response_build_stage_start = put_stage_metrics_enabled.then(Instant::now);
let mut response = S3Response::new(output);
// Echo XXHash3/64/128 / SHA-512 checksums that s3s PutObjectOutput has no typed
// field for (#1256).
inject_additional_checksum_headers(&mut response.headers, &committed.put_extra_checksum_headers);
rustfs_io_metrics::record_put_object_stage_duration_from("app_response_build", response_build_stage_start);
let result = Ok(response);
committed.finish(&result);
result
}
/// The single-object write path shared by the S3 handler and internal
/// callers: quota admission, foreground write admission, bucket default
/// SSE, Object Lock defaults, put options, the hashing/compressing/
/// encrypting reader, the owned store commit, usage accounting,
/// replication scheduling and the creation-event setup. The caller shapes
/// the request before and builds its response after.
pub(super) async fn put_object_core(
&self,
write: PutObjectWriteRequest<'_>,
body: StreamingBlob,
start_time: Instant,
) -> S3Result<PutObjectCommitted> {
let put_stage_metrics_enabled = rustfs_io_metrics::put_stage_metrics_enabled();
let PutObjectWriteRequest {
bucket,
key,
mut size,
quota_operation,
ciphertext_passthrough,
inbound_replication_put,
headers,
query,
trailing_headers,
version_id,
sse,
user_metadata,
internal_metadata,
content,
object_lock,
content_md5,
preserve_etag,
origin,
} = write;
let PutObjectSseInput {
server_side_encryption,
ssekms_key_id,
sse_customer_algorithm,
sse_customer_key,
sse_customer_key_md5,
} = sse;
// The app check preserves the existing S3 error contract; the storage
// commit path reserves the exact net logical growth under its locks.
let quota_stage_start = put_stage_metrics_enabled.then(Instant::now);
@@ -1084,7 +1438,7 @@ impl DefaultObjectUsecase {
let ingress_stage_start = put_stage_metrics_enabled.then(Instant::now);
let should_compress =
is_disk_compressible(&req.headers, &key) && size > MIN_DISK_COMPRESSIBLE_SIZE as i64 && !ciphertext_passthrough;
is_disk_compressible(headers, &key) && size > MIN_DISK_COMPRESSIBLE_SIZE as i64 && !ciphertext_passthrough;
// Resolve the store through the request-bound server context
// (backlog#1052 S6), not the process-global handle, so an embedded
@@ -1175,7 +1529,7 @@ impl DefaultObjectUsecase {
let (put_path, zero_copy_eager_put_path_status, use_zero_copy_eager_put_path, use_empty_or_small_eager_put_path) =
select_put_path_with_concurrency(
size,
&req.headers,
headers,
server_side_encryption_requested,
should_compress,
false,
@@ -1200,33 +1554,33 @@ impl DefaultObjectUsecase {
validate_sse_headers_for_write(
effective_sse.as_ref(),
effective_kms_key_id.as_ref(),
extract_ssekms_context_from_headers(&req.headers)?.as_ref(),
extract_ssekms_context_from_headers(headers)?.as_ref(),
sse_customer_algorithm.as_ref(),
sse_customer_key.as_ref(),
sse_customer_key_md5.as_ref(),
true, // PutObject requires all three: algorithm, key, key_md5
)?;
let mut metadata = metadata.unwrap_or_default();
let has_explicit_object_lock_retention = object_lock_mode.is_some()
|| object_lock_retain_until_date.is_some()
|| has_replication_retention_update(&req.headers, inbound_replication_put);
let mut metadata = user_metadata;
let has_explicit_object_lock_retention = object_lock.mode.is_some()
|| object_lock.retain_until_date.is_some()
|| has_replication_retention_update(headers, inbound_replication_put);
let object_lock_config_stage_start = put_stage_metrics_enabled.then(Instant::now);
let object_lock_config_state = load_bucket_object_lock_config_state(&bucket).await?;
rustfs_io_metrics::record_put_object_stage_duration_from("app_object_lock_config_lookup", object_lock_config_stage_start);
apply_put_request_metadata(
&mut metadata,
&req.headers,
headers,
&key,
cache_control,
content_disposition,
content_encoding,
content_language,
content_type,
expires,
website_redirect_location,
tagging,
storage_class.clone(),
content.cache_control,
content.content_disposition,
content.content_encoding,
content.content_language,
content.content_type,
content.expires,
content.website_redirect_location,
content.tagging,
content.storage_class,
)?;
apply_bucket_default_lock_retention(
&bucket,
@@ -1234,29 +1588,33 @@ impl DefaultObjectUsecase {
&mut metadata,
has_explicit_object_lock_retention,
)?;
metadata.extend(internal_metadata);
let put_opts_stage_start = put_stage_metrics_enabled.then(Instant::now);
let mut opts: ObjectOptions = put_opts_with_replication_authorization(
&bucket,
&key,
version_id.clone(),
&req.headers,
headers,
metadata.clone(),
replication_request_authorized(&req),
origin.replication_request_authorized(),
)
.await
.map_err(ApiError::from)?;
if let Some(etag) = preserve_etag {
opts.preserve_etag = Some(etag);
}
if let Some(quota_check) = quota_check.as_ref() {
apply_quota_admission(&mut opts, quota_check)?;
}
rustfs_io_metrics::record_put_object_stage_duration_from("app_put_opts_build", put_opts_stage_start);
apply_bucket_generation_guard(&req, &bucket, &mut opts)?;
origin.apply_bucket_generation_guard(&bucket, &mut opts)?;
apply_put_request_object_lock_opts(
&bucket,
&object_lock_config_state,
object_lock_legal_hold_status,
object_lock_mode,
object_lock_retain_until_date,
object_lock.legal_hold_status,
object_lock.mode,
object_lock.retain_until_date,
&mut opts,
)?;
let eager_put_commit_cancellation =
@@ -1278,7 +1636,7 @@ impl DefaultObjectUsecase {
let prelookup_stage_start = (prelookup_required && put_stage_metrics_enabled).then(Instant::now);
let prelookup_previous_current_size: Option<Option<u64>> = if prelookup_required {
let current_opts: ObjectOptions = internal_object_info_lookup_opts(
get_opts(&bucket, &key, version_id.clone(), None, &req.headers)
get_opts(&bucket, &key, version_id.clone(), None, headers)
.await
.map_err(ApiError::from)?,
);
@@ -1315,16 +1673,18 @@ impl DefaultObjectUsecase {
)?;
}
let mut md5hex = if let Some(base64_md5) = content_md5 {
let mut md5hex = match content_md5 {
Some(PutObjectContentMd5::Base64(base64_md5)) => {
let md5 = base64_simd::STANDARD
.decode_to_vec(base64_md5.as_bytes())
.map_err(|e| ApiError::from(StorageError::other(format!("Invalid content MD5: {e}"))))?;
Some(hex_simd::encode_to_string(&md5, hex_simd::AsciiCase::Lower))
} else {
None
}
Some(PutObjectContentMd5::Hex(md5hex)) => Some(md5hex),
None => None,
};
let mut sha256hex = get_content_sha256_with_query(&req.headers, req.uri.query());
let mut sha256hex = get_content_sha256_with_query(headers, query);
let mut write_plan = WritePlan::new();
// Additional-checksum (XXHash3/64/128, SHA-512) values to echo on the PutObject
@@ -1342,7 +1702,7 @@ impl DefaultObjectUsecase {
let mut hrd =
HashReader::from_stream(body, size, size, md5hex.take(), sha256hex.take(), false).map_err(ApiError::from)?;
if let Err(err) = hrd.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) {
if let Err(err) = hrd.add_checksum_from_s3s(headers, trailing_headers.clone(), false) {
return Err(ApiError::from(err).into());
}
@@ -1392,7 +1752,7 @@ impl DefaultObjectUsecase {
};
if size >= 0 {
if let Err(err) = reader.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) {
if let Err(err) = reader.add_checksum_from_s3s(headers, trailing_headers.clone(), false) {
return Err(ApiError::from(err).into());
}
@@ -1402,12 +1762,35 @@ impl DefaultObjectUsecase {
rustfs_io_metrics::record_put_object_path(put_path);
rustfs_io_metrics::record_put_object_stage_duration_from("ingress_prepare", ingress_stage_start);
let mut helper = OperationHelper::new(&req, event_name, S3Operation::PutObject);
let ssekms_context = extract_ssekms_context_from_headers(&req.headers)?;
let (mut completion, request_context) = match &origin {
PutObjectOrigin::S3 { req, event_name } => (
PutObjectCompletion::S3(OperationHelper::new(req, *event_name, S3Operation::PutObject)),
req.extensions.get::<request_context::RequestContext>().cloned(),
),
PutObjectOrigin::Internal {
principal_id,
emit_events,
} => {
let principal_id = *principal_id;
let request_context = request_context::RequestContext::fallback();
let event = emit_events.then(|| {
InternalPutObjectEvent::new(
current_notify_interface_for_context(self.context.as_deref()),
request_context.clone(),
EventName::ObjectCreatedPut,
&bucket,
&key,
principal_id,
)
});
(PutObjectCompletion::Internal(event.flatten().map(Box::new)), Some(request_context))
}
};
let ssekms_context = extract_ssekms_context_from_headers(headers)?;
// Apply encryption using unified SSE API.
let encryption_stage_start = put_stage_metrics_enabled.then(Instant::now);
let write_principal = SseKmsPrincipal::from_request(&req);
let write_principal = origin.sse_principal();
let encryption_request = EncryptionRequest {
bucket: &bucket,
key: &key,
@@ -1430,11 +1813,7 @@ impl DefaultObjectUsecase {
} else {
match sse_encryption(encryption_request).await {
Ok(material) => material,
Err(err) => {
let result = Err(err.into());
let _ = helper.complete(&result);
return result;
}
Err(err) => return Err(fail_put_object(completion, err.into())),
}
};
@@ -1460,7 +1839,6 @@ impl DefaultObjectUsecase {
let mt2 = metadata.clone();
opts.user_defined.extend(metadata);
let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
let request_id = request_context
.as_ref()
.map(|ctx| ctx.request_id.clone())
@@ -1660,100 +2038,41 @@ impl DefaultObjectUsecase {
let PutObjectCommitResult { obj_info, put_versioned } = match put_commit_result {
Ok(Ok(result)) => result,
Ok(Err(err)) => {
let result: S3Result<S3Response<PutObjectOutput>> = Err(err);
put_request_guard.finish_err();
let _ = helper.complete(&result);
return result;
return Err(fail_put_object(completion, err));
}
Err(err) => {
let result: S3Result<S3Response<PutObjectOutput>> = Err(S3Error::with_message(
S3ErrorCode::InternalError,
format!("put object commit owner task failed: {err}"),
));
put_request_guard.finish_err();
let _ = helper.complete(&result);
return result;
return Err(fail_put_object(
completion,
S3Error::with_message(S3ErrorCode::InternalError, format!("put object commit owner task failed: {err}")),
));
}
};
let raw_version = obj_info.version_id.map(|v| v.to_string());
helper = helper.object(obj_info.clone());
if let Some(version_id) = &raw_version {
helper = helper.version_id(version_id.clone());
completion = completion.object(obj_info.clone());
if let Some(version_id) = obj_info.version_id {
completion = completion.version_id(version_id.to_string());
}
let put_version = if put_versioned { raw_version } else { None };
let e_tag = obj_info.etag.clone().map(|etag| to_s3s_etag(&etag));
let expiration = resolve_put_object_expiration(&bucket, &obj_info).await;
let mut checksums = PutObjectChecksums {
crc32: input.checksum_crc32,
crc32c: input.checksum_crc32c,
sha1: input.checksum_sha1,
sha256: input.checksum_sha256,
crc64nvme: input.checksum_crc64nvme,
};
apply_trailing_checksums(
input.checksum_algorithm.as_ref().map(|a| a.as_str()),
&req.trailing_headers,
&mut checksums,
);
let output = PutObjectOutput {
e_tag,
server_side_encryption: effective_sse,
sse_customer_algorithm: sse_customer_algorithm.clone(),
sse_customer_key_md5: sse_customer_key_md5.clone(),
ssekms_key_id: effective_kms_key_id,
expiration,
checksum_crc32: checksums.crc32,
checksum_crc32c: checksums.crc32c,
checksum_sha1: checksums.sha1,
checksum_sha256: checksums.sha256,
checksum_crc64nvme: checksums.crc64nvme,
version_id: put_version,
..Default::default()
};
// For browser-based POST uploads (multipart/form-data), response status/body handling
// is decided by s3s PostObject serializer (success_action_status / redirect semantics).
let response_build_stage_start = put_stage_metrics_enabled.then(Instant::now);
let mut response = S3Response::new(output);
// Echo XXHash3/64/128 / SHA-512 checksums that s3s PutObjectOutput has no typed
// field for (#1256).
inject_additional_checksum_headers(&mut response.headers, &put_extra_checksum_headers);
rustfs_io_metrics::record_put_object_stage_duration_from("app_response_build", response_build_stage_start);
let result = Ok(response);
let _ = helper.complete(&result);
// Record PutObject metrics via zero-copy-metrics
{
let duration_ms = start_time.elapsed().as_millis() as f64;
rustfs_io_metrics::record_put_object(
duration_ms,
Ok(PutObjectCommitted {
obj_info,
put_versioned,
effective_sse,
effective_kms_key_id,
sse_customer_algorithm,
sse_customer_key_md5,
put_extra_checksum_headers,
completion,
put_request_guard,
bucket,
key,
start_time,
size,
use_zero_copy_eager_put_path, // Track if zero-copy was enabled
);
}
debug!(
target: "rustfs::app::object_usecase",
component = "app",
subsystem = "object",
bucket = %bucket,
key = %key,
use_zero_copy_eager_put_path,
concurrent_put_requests,
buffer_size,
"PutObject request completed"
);
put_request_guard.finish_ok();
result
})
}
}