From 183b5c9ede3dc04f3e49cb6c441c85a28808aee2 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 2 Sep 2026 22:56:23 +0800 Subject: [PATCH] refactor(object): extract internal put entry and ODM provenance keys (#7071) * feat(utils): add on-demand migration provenance metadata suffixes * refactor(object): extract internal put entry from the S3 PutObject path --- crates/utils/src/http/metadata_compat.rs | 89 +++ rustfs/src/app/object/internal_put.rs | 968 +++++++++++++++++++++++ rustfs/src/app/object/mod.rs | 5 + rustfs/src/app/object/put.rs | 573 +++++++++++--- 4 files changed, 1508 insertions(+), 127 deletions(-) create mode 100644 rustfs/src/app/object/internal_put.rs diff --git a/crates/utils/src/http/metadata_compat.rs b/crates/utils/src/http/metadata_compat.rs index 635c2dfb4..f4b2fe429 100644 --- a/crates/utils/src/http/metadata_compat.rs +++ b/crates/utils/src/http/metadata_compat.rs @@ -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 `:`. +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 = 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"; diff --git a/rustfs/src/app/object/internal_put.rs b/rustfs/src/app/object/internal_put.rs new file mode 100644 index 000000000..dcc406c1f --- /dev/null +++ b/rustfs/src/app/object/internal_put.rs @@ -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, + /// Lowercase hex MD5 the body must hash to; the write fails with + /// `BadDigest` otherwise and nothing is committed. + pub(crate) expected_md5_hex: Option, + /// ETag to store instead of the computed one. + pub(crate) preserve_etag: Option, + pub(crate) content_headers: HashMap, + pub(crate) user_metadata: HashMap, + pub(crate) tags: Option, + pub(crate) internal_metadata: HashMap, + /// 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) -> Result { + 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 { + headers.get(name).and_then(|value| value.to_str().ok()).map(ToOwned::to_owned) +} + +fn internal_put_content_input(headers: &HeaderMap, tags: Option) -> 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, + 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, + request_context: request_context::RequestContext, + event_name: EventName, + bucket: &str, + key: &str, + principal_id: &'static str, + ) -> Option { + 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(self, result: &S3Result>) { + 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(&self, ctx: InternalPutContext, body: B) -> Result + where + B: Stream> + 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> = 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 { + 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, ¤t_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( + &self, + ctx: &InternalPutContext, + upload_id: &str, + part_number: usize, + size: u64, + expected_md5_hex: Option, + body: B, + ) -> Result + where + B: Stream> + 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, + ) -> Result { + 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, ¤t_opts).await { + Ok(existing_obj_info) => { + validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, ¤t_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> = 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) -> impl Stream> + Send + Sync + Unpin + 'static { + futures::stream::iter(chunks.into_iter().map(Ok)) + } + + fn provenance_metadata() -> HashMap { + 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, 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); + } +} diff --git a/rustfs/src/app/object/mod.rs b/rustfs/src/app/object/mod.rs index a7bf0b93e..a5ca06d7e 100644 --- a/rustfs/src/app/object/mod.rs +++ b/rustfs/src/app/object/mod.rs @@ -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::*; diff --git a/rustfs/src/app/object/put.rs b/rustfs/src/app/object/put.rs index 74418f4af..8f506306a 100644 --- a/rustfs/src/app/object/put.rs +++ b/rustfs/src/app/object/put.rs @@ -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, + pub(super) content_disposition: Option, + pub(super) content_encoding: Option, + pub(super) content_language: Option, + pub(super) content_type: Option, + pub(super) expires: Option, + pub(super) website_redirect_location: Option, + pub(super) tagging: Option, + pub(super) storage_class: Option, +} + +/// Encryption inputs of a PUT after the S3 input/header merge. +pub(super) struct PutObjectSseInput { + pub(super) server_side_encryption: Option, + pub(super) ssekms_key_id: Option, + pub(super) sse_customer_algorithm: Option, + pub(super) sse_customer_key: Option, + pub(super) sse_customer_key_md5: Option, +} + +/// 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, + pub(super) mode: Option, + pub(super) retain_until_date: Option, +} + +/// 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, + 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 { + 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, + pub(super) version_id: Option, + pub(super) sse: PutObjectSseInput, + /// User metadata keyed the way s3s delivers it (`x-amz-meta-` stripped). + pub(super) user_metadata: HashMap, + /// Internal `x-rustfs-internal-*` / `x-minio-internal-*` keys written + /// verbatim onto the object; empty for S3 requests. + pub(super) internal_metadata: HashMap, + pub(super) content: PutObjectContentInput, + pub(super) object_lock: PutObjectLockInput, + pub(super) content_md5: Option, + /// ETag to store instead of the computed one; `None` keeps the computed + /// (or replication-header-derived) value. + pub(super) preserve_etag: Option, + pub(super) origin: PutObjectOrigin<'a>, +} + +/// Audit/notification completion of a write, per origin. +pub(super) enum PutObjectCompletion { + S3(OperationHelper), + Internal(Option>), +} + +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(self, result: &S3Result>) -> 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> = 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, + pub(super) effective_kms_key_id: Option, + pub(super) sse_customer_algorithm: Option, + pub(super) sse_customer_key_md5: Option, + 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(self, result: &S3Result>) { + 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 { + 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> = 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 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 + 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)) + } + 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::().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::().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> = 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> = 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, - 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, + 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, concurrent_put_requests, buffer_size, - "PutObject request completed" - ); - - put_request_guard.finish_ok(); - - result + }) } }