diff --git a/rustfs/src/app/object/extract.rs b/rustfs/src/app/object/extract.rs index 1fbbae649..12f367a13 100644 --- a/rustfs/src/app/object/extract.rs +++ b/rustfs/src/app/object/extract.rs @@ -15,6 +15,12 @@ //! Snowball auto-extract (PutObject x-amz-meta-snowball-auto-extract) path. use super::*; +use crate::app::storage_api::object_usecase::bucket::replication::ReplicateDecision; +#[cfg(test)] +use crate::app::storage_api::object_usecase::concurrency::SNOWBALL_MEMBER_COMMIT_LIMIT; +use crate::app::storage_api::object_usecase::concurrency::SNOWBALL_STAGING_BYTES_LIMIT; +use futures::stream::FuturesUnordered; +use std::collections::HashSet; // One logical member can be preceded by local PAX, GNU long-name, and GNU // long-link records. Count all four physical headers without rejecting that @@ -240,8 +246,12 @@ fn track_extract_member_read_errors(reader: HashReader) -> std::io::Result<(Hash Ok((tracked, failed)) } -fn should_ignore_extract_member_write_error(ignore_errors: bool, member_read_failed: &AtomicBool) -> bool { - ignore_errors && !member_read_failed.load(Ordering::Acquire) +fn classify_extract_member_write_error(error: S3Error, member_read_failed: &AtomicBool) -> ExtractCommitError { + if member_read_failed.load(Ordering::Acquire) { + ExtractCommitError::Fatal(error) + } else { + ExtractCommitError::StorageWrite(error) + } } pin_project! { @@ -340,8 +350,579 @@ const EXTRACT_MAX_EFFECTIVE_PAX_HEADER_BYTES: usize = 8 * 1024; const EXTRACT_MAX_EFFECTIVE_PAX_USER_METADATA_BYTES: usize = 2 * 1024; const EXTRACT_MAX_EFFECTIVE_PAX_FIELDS: usize = 4096; const EXTRACT_MAX_EXPANDED_PAX_METADATA_BYTES: u64 = 128 * 1024 * 1024; +const EXTRACT_SMALL_MEMBER_MAX_BYTES: usize = 64 * 1024; +const EXTRACT_DEFAULT_MAX_INFLIGHT: usize = 1; +const EXTRACT_BATCH_MAX_MEMBERS: usize = 16; +const EXTRACT_BATCH_MAX_STAGING_BYTES: usize = 2 * 1024 * 1024; +const EXTRACT_MEMBER_CONTEXT_OVERHEAD_BYTES: usize = 512; +const EXTRACT_METADATA_ENTRY_OVERHEAD_BYTES: usize = 64; +const ENV_RUSTFS_SNOWBALL_EXTRACT_MAX_INFLIGHT: &str = "RUSTFS_SNOWBALL_EXTRACT_MAX_INFLIGHT"; const TAR_TYPEFLAG_OFFSET: usize = 156; +fn put_object_extract_max_inflight() -> usize { + static MAX_INFLIGHT: OnceLock = OnceLock::new(); + *MAX_INFLIGHT.get_or_init(|| { + normalize_put_object_extract_max_inflight(rustfs_utils::get_env_usize( + ENV_RUSTFS_SNOWBALL_EXTRACT_MAX_INFLIGHT, + EXTRACT_DEFAULT_MAX_INFLIGHT, + )) + }) +} + +fn normalize_put_object_extract_max_inflight(value: usize) -> usize { + value.clamp(1, EXTRACT_BATCH_MAX_MEMBERS) +} + +fn select_put_object_extract_max_inflight(configured: usize, ignore_errors: bool, quota_enabled: bool) -> usize { + if ignore_errors && !quota_enabled { configured } else { 1 } +} + +struct ExtractPutRequestGuard { + inner: PutObjectGuard, + succeeded: bool, +} + +impl ExtractPutRequestGuard { + fn new() -> Self { + Self { + inner: PutObjectGuard::new(), + succeeded: false, + } + } + + fn finish_ok(&mut self) { + self.succeeded = true; + } +} + +impl Drop for ExtractPutRequestGuard { + fn drop(&mut self) { + if self.succeeded { + self.inner.finish_ok(); + } else { + self.inner.finish_err(); + } + } +} + +struct ExtractStagedBody { + reader: Option, +} + +impl ExtractStagedBody { + fn empty() -> Self { + Self { reader: None } + } +} + +impl AsyncRead for ExtractStagedBody { + fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, target: &mut ReadBuf<'_>) -> Poll> { + let Some(reader) = self.reader.as_mut() else { + return Poll::Ready(Ok(())); + }; + Pin::new(reader).poll_read(cx, target) + } +} + +async fn stage_extract_member_body(body: &mut R, size: usize) -> S3Result +where + R: AsyncRead + Unpin, +{ + if size == 0 { + return Ok(ExtractStagedBody::empty()); + } + + let pool = get_concurrency_manager().bytes_pool(); + let mut buffer = pool.acquire_buffer(size).await; + super::put::read_small_put_body_into(body, &mut *buffer, size).await?; + Ok(ExtractStagedBody { + reader: Some(super::put::PooledBufferReader::new(buffer, size)), + }) +} + +fn checked_extract_hash_map_allocation(map: &HashMap) -> Option { + // HashMap does not expose its raw bucket allocation. Twice the advertised + // element capacity conservatively covers buckets plus control bytes while + // keeping the accounting independent of the current hashbrown layout. + map.capacity() + .checked_mul(2)? + .checked_mul(std::mem::size_of::<(K, V)>().checked_add(1)?) +} + +fn checked_extract_member_staging_weight( + path: &str, + size: usize, + opts: &ObjectOptions, + replication: &ReplicateDecision, +) -> S3Result { + let body_reservation = if size == 0 { 0 } else { EXTRACT_SMALL_MEMBER_MAX_BYTES }; + // The authorization request is dropped before staging. Account the two + // retained key copies plus all dynamic maps frozen into ObjectOptions; + // the fixed allowance covers the remaining options/write-plan handles. + let mut total = body_reservation + .checked_add( + path.len() + .checked_mul(2) + .ok_or_else(|| object_s3_error(S3ErrorCode::InvalidArgument, "Snowball prepared member path size overflowed"))?, + ) + .and_then(|bytes| bytes.checked_add(EXTRACT_MEMBER_CONTEXT_OVERHEAD_BYTES)) + .ok_or_else(|| object_s3_error(S3ErrorCode::InvalidArgument, "Snowball prepared member size overflowed"))?; + for metadata in std::iter::once(&opts.user_defined).chain(opts.eval_metadata.iter()) { + for (name, value) in metadata { + total = total + .checked_add(name.len()) + .and_then(|bytes| bytes.checked_add(value.len())) + .and_then(|bytes| bytes.checked_add(EXTRACT_METADATA_ENTRY_OVERHEAD_BYTES)) + .ok_or_else(|| object_s3_error(S3ErrorCode::InvalidArgument, "Snowball prepared member size overflowed"))?; + } + } + total = total + .checked_add( + checked_extract_hash_map_allocation(&replication.targets_map) + .ok_or_else(|| object_s3_error(S3ErrorCode::InvalidArgument, "Snowball replication decision size overflowed"))?, + ) + .ok_or_else(|| object_s3_error(S3ErrorCode::InvalidArgument, "Snowball prepared member size overflowed"))?; + for (target_name, target) in &replication.targets_map { + total = total + .checked_add(target_name.capacity()) + .and_then(|bytes| bytes.checked_add(target.arn.capacity())) + .and_then(|bytes| bytes.checked_add(target.id.capacity())) + .ok_or_else(|| object_s3_error(S3ErrorCode::InvalidArgument, "Snowball replication decision size overflowed"))?; + } + Ok(total) +} + +fn try_acquire_extract_staging_permit(manager: &ConcurrencyManager, staging_weight: usize) -> S3Result { + if staging_weight > SNOWBALL_STAGING_BYTES_LIMIT { + return Err(object_s3_error( + S3ErrorCode::SlowDown, + "Snowball member retained state exceeds the global staging budget", + )); + } + let permits = u32::try_from(staging_weight).map_err(|_| { + object_s3_error(S3ErrorCode::SlowDown, "Snowball member retained state exceeds the global staging budget") + })?; + manager.try_acquire_snowball_staging_bytes(permits).ok_or_else(|| { + object_s3_error( + S3ErrorCode::SlowDown, + "Snowball staging budget is exhausted, please reduce your request rate", + ) + }) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ExtractBatchAction { + Stage, + FlushThenStage, + SerialBarrier, +} + +#[derive(Debug, Default)] +struct ExtractBatchState { + keys: HashSet, + staging_bytes: usize, +} + +impl ExtractBatchState { + fn action( + &self, + key: &str, + member_size: usize, + staging_weight: usize, + max_inflight: usize, + durable_quota: bool, + ) -> ExtractBatchAction { + if durable_quota + || max_inflight <= 1 + || member_size > EXTRACT_SMALL_MEMBER_MAX_BYTES + || staging_weight > EXTRACT_BATCH_MAX_STAGING_BYTES + { + return ExtractBatchAction::SerialBarrier; + } + + if self.keys.contains(key) + || self.keys.len() >= max_inflight + || self + .staging_bytes + .checked_add(staging_weight) + .is_none_or(|bytes| bytes > EXTRACT_BATCH_MAX_STAGING_BYTES) + { + ExtractBatchAction::FlushThenStage + } else { + ExtractBatchAction::Stage + } + } + + fn record(&mut self, key: &str, staging_weight: usize) { + debug_assert!(!self.keys.contains(key)); + self.keys.insert(key.to_string()); + self.staging_bytes = self.staging_bytes.saturating_add(staging_weight); + } + + fn clear(&mut self) { + self.keys.clear(); + self.staging_bytes = 0; + } +} + +struct ExtractPreparedMember { + archive_seq: usize, + key: String, + size: i64, + actual_size: i64, + body: R, + write_plan: WritePlan, + opts: ObjectOptions, + replication: ReplicateDecision, + staging_permit: OwnedSemaphorePermit, + member_permit: OwnedSemaphorePermit, +} + +struct ExtractCommitContext { + store: Arc, + cache_adapter: Arc, + notify: Arc, + bucket: String, + quota_enabled: bool, + request_context: request_context::RequestContext, + req_params: hashbrown::HashMap, + host: String, + port: u16, + user_agent: String, + wrote_any_entry: AtomicBool, +} + +struct ExtractCommitSuccess { + event: rustfs_notify::EventArgs, + post_commit_error: Option, +} + +enum ExtractCommitError { + StorageWrite(S3Error), + Fatal(S3Error), +} + +impl From for ExtractCommitError { + fn from(error: ApiError) -> Self { + Self::Fatal(error.into()) + } +} + +impl ExtractCommitError { + fn into_unignored(self, ignore_errors: bool) -> Option { + match self { + Self::StorageWrite(error) if ignore_errors => { + warn!(error = %error, "Archive object write skipped due to ignore-errors"); + None + } + Self::StorageWrite(error) | Self::Fatal(error) => Some(error), + } + } +} + +struct ExtractCommitOutcome { + archive_seq: usize, + event: Option, + error: Option, +} + +async fn run_extract_owned_task(task: F) -> S3Result +where + T: Send + 'static, + F: std::future::Future + Send + 'static, +{ + spawn_traced_join(task) + .await + .map_err(|error| S3Error::with_message(S3ErrorCode::InternalError, format!("Snowball commit owner task failed: {error}"))) +} + +async fn complete_extract_member_post_commit( + context: Arc, + key: String, + opts: ObjectOptions, + replication: ReplicateDecision, + obj_info: ObjectInfo, + backfilled_old_current_size: Option, +) -> ExtractCommitSuccess { + let extract_versioned = opts.versioned; + let post_commit_error = match quota_accounting_object_size(&obj_info, context.quota_enabled) { + Ok(committed_size) => { + match previous_current_size_from_backfill(backfilled_old_current_size) { + Some(previous_current_size) => { + if extract_versioned { + record_bucket_object_version_write_memory(&context.bucket, previous_current_size, committed_size).await; + } else { + record_bucket_object_write_memory(&context.bucket, previous_current_size, committed_size).await; + } + } + None => { + record_bucket_object_write_unknown_previous_memory(&context.bucket, committed_size, extract_versioned).await; + } + } + None + } + Err(err) => Some(err), + }; + let _ = invalidate_object_data_cache_after_put_success(&context.cache_adapter, &context.bucket, &key).await; + + if replication.replicate_any() { + schedule_object_replication(obj_info.clone(), context.store.clone(), replication).await; + } + + let e_tag = obj_info.etag.clone().map(|etag| to_s3s_etag(&etag)); + let output = PutObjectOutput { + e_tag, + ..Default::default() + }; + let actual_version_id = extract_notification_version_id(obj_info.version_id, opts.versioned, opts.version_suspended); + let mut event_object = convert_ecstore_object_info(obj_info); + event_object.version_id = (!actual_version_id.is_empty()).then_some(actual_version_id.clone()); + let event = rustfs_notify::EventArgs { + event_name: put_event_name_for_post_object(false), + bucket_name: context.bucket.clone(), + object: event_object, + req_params: context.req_params.clone(), + resp_elements: build_event_resp_elements(&S3Response::new(output), &context.request_context.request_id), + version_id: actual_version_id, + host: context.host.clone(), + port: context.port, + user_agent: context.user_agent.clone(), + }; + + ExtractCommitSuccess { + event, + post_commit_error, + } +} + +async fn commit_extract_member_inner( + context: Arc, + member: ExtractPreparedMember, + foreground_permit: Option, +) -> Result +where + R: AsyncRead + Send + Sync + Unpin + 'static, +{ + let ExtractPreparedMember { + archive_seq: _, + key, + size, + actual_size, + body, + write_plan, + opts, + replication, + staging_permit, + member_permit, + } = member; + let hrd = HashReader::from_stream(body, size, actual_size, None, None, false).map_err(ApiError::from)?; + let hrd = write_plan.apply(hrd, actual_size).map_err(ApiError::from)?; + let (hrd, member_read_failed) = track_extract_member_read_errors(hrd).map_err(ApiError::from)?; + let mut reader = PutObjReader::new(hrd); + let _ = invalidate_object_data_cache_before_mutation(&context.cache_adapter, &context.bucket, &key).await; + + let (obj_info, backfilled_old_current_size) = match context + .store + .put_object_with_old_current_size(&context.bucket, &key, &mut reader, &opts) + .await + { + Ok(result) => result, + Err(error) => { + let error: S3Error = ApiError::from(error).into(); + return Err(classify_extract_member_write_error(error, &member_read_failed)); + } + }; + drop(reader); + drop(staging_permit); + drop(foreground_permit); + + // The independently owned commit publishes the authoritative mutation to + // the scanner before its post-store awaits, then retains the lifecycle slot + // through quota, cache, replication, and event construction. + if !context.wrote_any_entry.swap(true, Ordering::AcqRel) { + rustfs_scanner::record_dirty_usage_bucket(&context.bucket); + } + let success = + complete_extract_member_post_commit(context, key, opts, replication, obj_info, backfilled_old_current_size).await; + drop(member_permit); + Ok(success) +} + +async fn commit_extract_member(context: Arc, member: ExtractPreparedMember) -> ExtractCommitOutcome +where + R: AsyncRead + Send + Sync + Unpin + 'static, +{ + let archive_seq = member.archive_seq; + let manager = get_concurrency_manager(); + let foreground_permit = match manager.admit_snowball_foreground_write(member.actual_size).await { + Ok(ForegroundWriteAdmission::Disabled) => None, + Ok(ForegroundWriteAdmission::Admitted(permit)) => Some(permit), + Ok(ForegroundWriteAdmission::Rejected) => { + return ExtractCommitOutcome { + archive_seq, + event: None, + error: Some(ExtractCommitError::StorageWrite(object_s3_error( + S3ErrorCode::SlowDown, + "foreground write concurrency limit reached, please reduce your request rate", + ))), + }; + } + Err(_) => { + return ExtractCommitOutcome { + archive_seq, + event: None, + error: Some(ExtractCommitError::Fatal(object_s3_error( + S3ErrorCode::InternalError, + "Snowball foreground write admission closed", + ))), + }; + } + }; + match commit_extract_member_inner(context, member, foreground_permit).await { + Ok(success) => ExtractCommitOutcome { + archive_seq, + event: Some(success.event), + error: success.post_commit_error.map(ExtractCommitError::Fatal), + }, + Err(error) => ExtractCommitOutcome { + archive_seq, + event: None, + error: Some(error), + }, + } +} + +fn ordered_extract_outcomes( + mut outcomes: Vec, + ignore_errors: bool, +) -> (Vec, Option) { + outcomes.sort_by_key(|outcome| outcome.archive_seq); + let mut events = Vec::with_capacity(outcomes.len()); + let mut earliest_error = None; + for outcome in outcomes { + if let Some(event) = outcome.event { + events.push(event); + } + let error = outcome.error.and_then(|error| error.into_unignored(ignore_errors)); + if earliest_error.is_none() { + earliest_error = error; + } + } + (events, earliest_error) +} + +fn spawn_extract_notification_batch( + request_context: Option, + events: Vec, + notify: F, +) where + F: Fn(rustfs_notify::EventArgs) -> Fut + Send + Sync + 'static, + Fut: std::future::Future + Send + 'static, +{ + spawn_background_with_context(request_context, async move { + for event in events { + notify(event).await; + } + }); +} + +fn finish_extract_outcomes( + outcomes: Vec, + context: &Arc, + ignore_errors: bool, +) -> S3Result<()> { + let (events, earliest_error) = ordered_extract_outcomes(outcomes, ignore_errors); + if !events.is_empty() { + let notify = context.notify.clone(); + spawn_extract_notification_batch(Some(context.request_context.clone()), events, move |event| { + let notify = notify.clone(); + async move { + notify.notify(event).await; + } + }); + } + match earliest_error { + Some(error) => Err(error), + None => Ok(()), + } +} + +async fn run_extract_outcomes_owner(context: Arc, ignore_errors: bool, commits: F) -> S3Result<()> +where + F: std::future::Future> + Send + 'static, +{ + run_extract_owned_task(async move { + let outcomes = commits.await; + finish_extract_outcomes(outcomes, &context, ignore_errors) + }) + .await? +} + +async fn drain_extract_commits(mut commits: FuturesUnordered) -> Vec +where + F: std::future::Future, +{ + let mut outcomes = Vec::with_capacity(commits.len()); + while let Some(outcome) = commits.next().await { + outcomes.push(outcome); + } + outcomes +} + +async fn run_extract_commits(members: impl IntoIterator, mut commit: F) -> Vec +where + F: FnMut(T) -> Fut, + Fut: std::future::Future, +{ + let commits = FuturesUnordered::new(); + for member in members { + commits.push(commit(member)); + } + drain_extract_commits(commits).await +} + +async fn flush_extract_batch( + batch: &mut Vec, + batch_state: &mut ExtractBatchState, + context: &Arc, + ignore_errors: bool, +) -> S3Result<()> { + if batch.is_empty() { + batch_state.clear(); + return Ok(()); + } + + let members = batch.split_off(0); + batch_state.clear(); + let commit_context = context.clone(); + run_extract_outcomes_owner(context.clone(), ignore_errors, async move { + run_extract_commits(members, |member| commit_extract_member(commit_context.clone(), member)).await + }) + .await +} + +async fn acquire_extract_member_lifecycle_permit( + batch: &mut Vec, + batch_state: &mut ExtractBatchState, + context: &Arc, + ignore_errors: bool, +) -> S3Result { + let manager = get_concurrency_manager(); + if let Some(permit) = manager.try_acquire_snowball_member_commit() { + return Ok(permit); + } + if !batch.is_empty() { + flush_extract_batch(batch, batch_state, context, ignore_errors).await?; + if let Some(permit) = manager.try_acquire_snowball_member_commit() { + return Ok(permit); + } + } + manager + .acquire_snowball_member_commit() + .await + .map_err(|_| object_s3_error(S3ErrorCode::InternalError, "Snowball member lifecycle admission closed")) +} + #[derive(Debug, Clone, Default, PartialEq, Eq)] struct PutObjectExtractOptions { prefix: Option, @@ -1433,6 +2014,27 @@ impl DefaultObjectUsecase { u64::try_from(size).map_err(|_| S3Error::new(S3ErrorCode::UnexpectedContent))?, ) .await?; + match get_concurrency_manager() + .admit_put_object(size) + .await + .map_err(|_| object_s3_error(S3ErrorCode::InternalError, "foreground write admission closed"))? + { + ForegroundWriteAdmission::Disabled => {} + ForegroundWriteAdmission::Admitted(permit) => { + counter!("rustfs.put_object.foreground_admission.total", "result" => "admitted").increment(1); + // The archive admission is a preflight. Each member later uses + // its own logical size against the same foreground PUT policy. + drop(permit); + } + ForegroundWriteAdmission::Rejected => { + counter!("rustfs.put_object.foreground_admission.total", "result" => "rejected").increment(1); + return Err(object_s3_error( + S3ErrorCode::SlowDown, + "foreground write concurrency limit reached, please reduce your request rate", + )); + } + } + let mut put_request_guard = ExtractPutRequestGuard::new(); // Apply adaptive buffer sizing based on file size for optimal streaming performance. // Uses workload profile configuration (enabled by default) to select appropriate buffer size. @@ -1503,7 +2105,6 @@ impl DefaultObjectUsecase { let host = get_request_host(&req.headers); let port = get_request_port(&req.headers); let user_agent = get_request_user_agent(&req.headers); - let mut wrote_any_entry = false; let mut extracted_entry_count = 0usize; let mut resource_total_size = 0u64; let mut legacy_quota_growth = 0u64; @@ -1512,31 +2113,78 @@ impl DefaultObjectUsecase { let mut total_expanded_pax_metadata = 0u64; let object_lock_config_snapshot = store.object_lock_config_snapshot(&bucket).await.map_err(ApiError::from)?; let object_lock_config_state = object_lock_config_snapshot.state(); + let commit_context = Arc::new(ExtractCommitContext { + store, + cache_adapter: self.object_data_cache(), + notify, + bucket: bucket.clone(), + quota_enabled: extract_quota_enabled, + request_context, + req_params, + host, + port, + user_agent, + wrote_any_entry: AtomicBool::new(false), + }); + let durable_quota = extract_quota_check + .as_ref() + .is_some_and(|result| result.uses_durable_reservations); + // Without ignore-errors, the legacy contract stops before attempting a + // later member after the first storage failure. Parallel commits cannot + // preserve that boundary, so concurrency requires both ignore-errors + // and an explicit max-inflight value above the serial default. Quota + // accounting can fail after storage commit, so quota-enabled imports + // also remain serial. An opted-in micro-batch is always drained; a + // fatal outcome stops later batches but cannot roll back peers that + // already committed in the current batch. + let max_inflight = select_put_object_extract_max_inflight( + put_object_extract_max_inflight(), + extract_options.ignore_errors, + extract_quota_enabled, + ); + let mut batch = Vec::with_capacity(max_inflight); + let mut batch_state = ExtractBatchState::default(); + + macro_rules! extract_try { + ($expression:expr) => { + match $expression { + Ok(value) => value, + Err(error) => { + let error: S3Error = error.into(); + flush_extract_batch(&mut batch, &mut batch_state, &commit_context, extract_options.ignore_errors).await?; + return Err(error); + } + } + }; + } while let Some(entry) = entries.next().await { let mut f = match entry { Ok(f) => f, - Err(e) => { - error!(error = %e, "Archive entry read failed"); - return Err(s3_error!(InvalidArgument, "Failed to read archive entry: {:?}", e)); + Err(error) => { + error!(error = %error, "Archive entry read failed"); + flush_extract_batch(&mut batch, &mut batch_state, &commit_context, extract_options.ignore_errors).await?; + return Err(s3_error!(InvalidArgument, "Failed to read archive entry: {:?}", error)); } }; extracted_entry_count = extracted_entry_count.saturating_add(1); - validate_put_object_extract_entry_count(extracted_entry_count, extract_limits)?; + extract_try!(validate_put_object_extract_entry_count(extracted_entry_count, extract_limits,)); let entry_size = f.effective_size(); - validate_put_object_extract_entry_size("archive member", entry_size, extract_limits)?; - resource_total_size = resource_total_size - .checked_add(entry_size) - .ok_or_else(|| s3_error!(InvalidArgument, "Archive total unpacked size overflowed while processing entries"))?; - validate_put_object_extract_total_size(resource_total_size, extract_limits)?; - count_extract_entry_pax_metadata( - &mut f, - &mut total_pax_metadata_size, - &mut total_pax_metadata_records, - extract_limits, - ) - .await - .map_err(ExtractEntryError::into_s3_error)?; + extract_try!(validate_put_object_extract_entry_size("archive member", entry_size, extract_limits,)); + resource_total_size = extract_try!(resource_total_size.checked_add(entry_size).ok_or_else(|| { + s3_error!(InvalidArgument, "Archive total unpacked size overflowed while processing entries") + })); + extract_try!(validate_put_object_extract_total_size(resource_total_size, extract_limits,)); + extract_try!( + count_extract_entry_pax_metadata( + &mut f, + &mut total_pax_metadata_size, + &mut total_pax_metadata_records, + extract_limits, + ) + .await + .map_err(ExtractEntryError::into_s3_error) + ); let archive_entry_type = f.header().entry_type(); let entry_kind = classify_extract_entry_type(archive_entry_type); @@ -1544,14 +2192,14 @@ impl DefaultObjectUsecase { ExtractEntryKind::Skip => continue, ExtractEntryKind::Directory | ExtractEntryKind::Object => {} } - validate_extract_special_entry_size(archive_entry_type, entry_size)?; + extract_try!(validate_extract_special_entry_size(archive_entry_type, entry_size)); let (fpath, is_dir) = { - let path_bytes = f.path_bytes().map_err(map_extract_archive_error)?; + let path_bytes = extract_try!(f.path_bytes().map_err(map_extract_archive_error)); let path = match strict_extract_entry_path(path_bytes.as_ref()) { Ok(path) => path, - Err(err) => { - err.ignore_or_return(extract_options.ignore_errors)?; + Err(error) => { + extract_try!(error.ignore_or_return(extract_options.ignore_errors)); continue; } }; @@ -1564,19 +2212,28 @@ impl DefaultObjectUsecase { } let fpath = match normalize_extract_entry_key(path, extract_options.prefix.as_deref(), is_dir) { Ok(fpath) => fpath, - Err(err) => { - ExtractEntryError::Recoverable(err).ignore_or_return(extract_options.ignore_errors)?; + Err(error) => { + extract_try!(ExtractEntryError::Fatal(error).ignore_or_return(extract_options.ignore_errors)); continue; } }; (fpath, is_dir) }; - if let Err(err) = validate_extract_member_key(&fpath, extract_limits) { - err.ignore_or_return(extract_options.ignore_errors)?; + if let Err(error) = validate_extract_member_key(&fpath, extract_limits) { + extract_try!(error.ignore_or_return(extract_options.ignore_errors)); continue; } - validate_table_catalog_object_mutation(&bucket, &fpath).await?; + if batch_state.keys.contains(&fpath) + || durable_quota + || max_inflight <= 1 + || (!is_dir && entry_size > EXTRACT_SMALL_MEMBER_MAX_BYTES as u64) + { + extract_try!( + flush_extract_batch(&mut batch, &mut batch_state, &commit_context, extract_options.ignore_errors,).await + ); + } + extract_try!(validate_table_catalog_object_mutation(&bucket, &fpath).await); let mut auth_req = S3Request { input: PutObjectInput::default(), @@ -1590,20 +2247,21 @@ impl DefaultObjectUsecase { trailing_headers: auth_trailing_headers.clone(), }; { - let req_info = req_info_mut(&mut auth_req)?; + let req_info = extract_try!(req_info_mut(&mut auth_req)); req_info.bucket = Some(bucket.clone()); req_info.object = Some(fpath.clone()); req_info.version_id = None; } - let mut size = - i64::try_from(entry_size).map_err(|_| s3_error!(InvalidArgument, "Archive entry size does not fit into i64"))?; + let mut size = extract_try!( + i64::try_from(entry_size).map_err(|_| s3_error!(InvalidArgument, "Archive entry size does not fit into i64")) + ); // mtime 0 or a negative GNU base-256 value means "unset". xl.meta // also cannot represent the Unix epoch as an object mod_time, so // those cases fall back to the upload time (rustfs#4842). - let archive_entry_mod_time = extract_archive_entry_mod_time(f.header())?; + let archive_entry_mod_time = extract_try!(extract_archive_entry_mod_time(f.header())); let mut metadata = HashMap::new(); let has_explicit_object_lock_retention = object_lock_mode.is_some() || object_lock_retain_until_date.is_some(); - apply_put_request_metadata( + extract_try!(apply_put_request_metadata( &mut metadata, &member_request_headers, &fpath, @@ -1616,25 +2274,27 @@ impl DefaultObjectUsecase { website_redirect_location.clone(), tagging.clone(), storage_class.clone(), - )?; - apply_bucket_default_lock_retention( + )); + extract_try!(apply_bucket_default_lock_retention( &bucket, object_lock_config_state, &mut metadata, has_explicit_object_lock_retention, - )?; - let mut opts = put_opts_with_replication_authorization( - &bucket, - &fpath, - outer_version_id.clone(), - &member_request_headers, - metadata.clone(), - replication_authorized, - ) - .await - .map_err(ApiError::from)?; + )); + let mut opts = extract_try!( + put_opts_with_replication_authorization( + &bucket, + &fpath, + outer_version_id.clone(), + &member_request_headers, + metadata.clone(), + replication_authorized, + ) + .await + .map_err(ApiError::from) + ); if let Some(quota_check) = extract_quota_check.as_ref() { - apply_quota_admission(&mut opts, quota_check)?; + extract_try!(apply_quota_admission(&mut opts, quota_check)); } opts.expected_bucket_incarnation_id = expected_bucket_incarnation_id; opts.object_lock_config_snapshot = Some(Arc::clone(&object_lock_config_snapshot)); @@ -1649,37 +2309,48 @@ impl DefaultObjectUsecase { .await { Ok(authorization) => authorization, - Err(err) => { - err.ignore_or_return(extract_options.ignore_errors)?; + Err(error) => { + extract_try!(error.ignore_or_return(extract_options.ignore_errors)); continue; } }; - total_expanded_pax_metadata = total_expanded_pax_metadata - .checked_add(pax_authorization.expanded_metadata_bytes) - .ok_or_else(|| object_s3_error(S3ErrorCode::InvalidArgument, "Snowball expanded PAX metadata size overflowed"))?; - validate_extract_expanded_pax_metadata_total(total_expanded_pax_metadata)?; - if let Some(quota_check) = extract_quota_check.as_ref() { - let next_legacy_quota_growth = legacy_quota_growth - .checked_add(extract_entry_quota_growth( - if is_dir { ExtractEntryKind::Directory } else { entry_kind }, - entry_size, - )) + total_expanded_pax_metadata = extract_try!( + total_expanded_pax_metadata + .checked_add(pax_authorization.expanded_metadata_bytes) .ok_or_else(|| { - object_s3_error(S3ErrorCode::InvalidArgument, "Archive quota growth overflowed while processing entries") - })?; - ensure_legacy_archive_size_within_quota(quota_check, next_legacy_quota_growth)?; + object_s3_error(S3ErrorCode::InvalidArgument, "Snowball expanded PAX metadata size overflowed") + }) + ); + extract_try!(validate_extract_expanded_pax_metadata_total(total_expanded_pax_metadata,)); + if let Some(quota_check) = extract_quota_check.as_ref() { + let next_legacy_quota_growth = extract_try!( + legacy_quota_growth + .checked_add(extract_entry_quota_growth( + if is_dir { ExtractEntryKind::Directory } else { entry_kind }, + entry_size, + )) + .ok_or_else(|| { + object_s3_error( + S3ErrorCode::InvalidArgument, + "Archive quota growth overflowed while processing entries", + ) + }) + ); + extract_try!(ensure_legacy_archive_size_within_quota(quota_check, next_legacy_quota_growth,)); legacy_quota_growth = next_legacy_quota_growth; } for (name, value) in &pax_authorization.headers { - auth_req.headers.try_insert(name.clone(), value.clone()).map_err(|_| { + extract_try!(auth_req.headers.try_insert(name.clone(), value.clone()).map_err(|_| { object_s3_error(S3ErrorCode::InvalidArgument, "Snowball IAM condition header capacity exceeded") - })?; + })); } let explicit_version_id = pax_authorization.version_id.as_deref().or(outer_version_id.as_deref()); - let authorization_version_id = explicit_version_id - .map(|version_id| apply_extract_version_id(version_id, &mut opts)) - .transpose()?; - req_info_mut(&mut auth_req)?.version_id = authorization_version_id; + let authorization_version_id = extract_try!( + explicit_version_id + .map(|version_id| apply_extract_version_id(version_id, &mut opts)) + .transpose() + ); + extract_try!(req_info_mut(&mut auth_req)).version_id = authorization_version_id; let effective_object_lock_legal_hold_status = pax_authorization .object_lock_legal_hold_status .clone() @@ -1701,19 +2372,20 @@ impl DefaultObjectUsecase { explicit_version_id, opts.delete_marker_replication_status() == ReplicationStatusType::Replica, ); - authorize_request(&mut auth_req, Action::S3Action(S3Action::PutObjectAction)).await?; + extract_try!(authorize_request(&mut auth_req, Action::S3Action(S3Action::PutObjectAction)).await); if iam_requirements.tagging { - authorize_request(&mut auth_req, Action::S3Action(S3Action::PutObjectTaggingAction)).await?; + extract_try!(authorize_request(&mut auth_req, Action::S3Action(S3Action::PutObjectTaggingAction)).await); } if iam_requirements.retention { - authorize_request(&mut auth_req, Action::S3Action(S3Action::PutObjectRetentionAction)).await?; + extract_try!(authorize_request(&mut auth_req, Action::S3Action(S3Action::PutObjectRetentionAction)).await); } if iam_requirements.legal_hold { - authorize_request(&mut auth_req, Action::S3Action(S3Action::PutObjectLegalHoldAction)).await?; + extract_try!(authorize_request(&mut auth_req, Action::S3Action(S3Action::PutObjectLegalHoldAction)).await); } if iam_requirements.replication { - authorize_request(&mut auth_req, Action::S3Action(S3Action::ReplicateObjectAction)).await?; + extract_try!(authorize_request(&mut auth_req, Action::S3Action(S3Action::ReplicateObjectAction)).await); } + drop(auth_req); if archive_entry_mod_time.is_some() { opts.mod_time = archive_entry_mod_time; } @@ -1723,69 +2395,65 @@ impl DefaultObjectUsecase { if is_dir { size = 0; } - let actual_size = size; - let should_compress = !is_dir && is_disk_compressible(&HeaderMap::new(), &fpath) && size > MIN_DISK_COMPRESSIBLE_SIZE as i64; - let mut write_plan = WritePlan::new(); - let mut hrd = if is_dir { - HashReader::from_stream(std::io::Cursor::new(Vec::new()), size, actual_size, None, None, false) - .map_err(ApiError::from)? - } else if should_compress { + if should_compress { let algorithm = CompressionAlgorithm::default(); insert_str(&mut metadata, SUFFIX_COMPRESSION, compression_metadata_value(algorithm)); insert_str(&mut metadata, SUFFIX_ACTUAL_SIZE, size.to_string()); - - let hrd = HashReader::from_stream(f, size, actual_size, None, None, false).map_err(ApiError::from)?; write_plan = write_plan.with_compression(algorithm); - hrd - } else { - HashReader::from_stream(f, size, actual_size, None, None, false).map_err(ApiError::from)? - }; - apply_put_request_object_lock_opts( + } + extract_try!(apply_put_request_object_lock_opts( &bucket, object_lock_config_state, effective_object_lock_legal_hold_status, effective_object_lock_mode, effective_object_lock_retain_until_date, &mut opts, - )?; - if let Some(material) = sse_encryption(EncryptionRequest { - bucket: &bucket, - key: &fpath, - server_side_encryption: effective_sse.clone(), - ssekms_key_id: effective_kms_key_id.clone(), - ssekms_context: extract_ssekms_context_from_headers(&req.headers)?, - sse_customer_algorithm: sse_customer_algorithm.clone(), - sse_customer_key: sse_customer_key.clone(), - sse_customer_key_md5: sse_customer_key_md5.clone(), - content_size: actual_size, - principal: extract_principal.as_ref(), - }) - .await? - { + )); + if let Some(material) = extract_try!( + sse_encryption(EncryptionRequest { + bucket: &bucket, + key: &fpath, + server_side_encryption: effective_sse.clone(), + ssekms_key_id: effective_kms_key_id.clone(), + ssekms_context: extract_try!(extract_ssekms_context_from_headers(&req.headers)), + sse_customer_algorithm: sse_customer_algorithm.clone(), + sse_customer_key: sse_customer_key.clone(), + sse_customer_key_md5: sse_customer_key_md5.clone(), + content_size: actual_size, + principal: extract_principal.as_ref(), + }) + .await + ) { effective_sse = Some(material.server_side_encryption.clone()); effective_kms_key_id = material.kms_key_id.clone(); - write_plan = write_plan.with_encryption(material.write_encryption(None)); - - let encryption_metadata = encryption_material_to_metadata(&material)?; + let encryption_metadata = extract_try!(encryption_material_to_metadata(&material)); metadata.extend(encryption_metadata.clone()); opts.user_defined.extend(encryption_metadata); } - hrd = write_plan.apply(hrd, actual_size).map_err(ApiError::from)?; - let (hrd, member_read_failed) = track_extract_member_read_errors(hrd).map_err(ApiError::from)?; opts.user_defined.extend(metadata); - // Each extracted member is an independent user write and joins - // bucket replication like a regular PUT (MinIO PutObjectExtract - // parity). One immutable decision drives both the pending metadata - // and the post-commit schedule below, same contract as the PUT path - // (https://github.com/rustfs/backlog/issues/1320); inbound replica - // writes are declined inside `must_replicate_object`. - let dsc = must_replicate_object( + // Reserve the global member lifecycle before constructing the + // potentially large replication decision. If this archive already + // owns a batch, drain it before waiting so requests cannot deadlock + // while each retains lifecycle capacity needed by its own batch. + let member_permit = extract_try!( + acquire_extract_member_lifecycle_permit( + &mut batch, + &mut batch_state, + &commit_context, + extract_options.ignore_errors, + ) + .await + ); + + // One immutable decision drives both the pending metadata and the + // post-commit schedule, matching the regular PUT contract. + let replication = must_replicate_object( &bucket, &fpath, &opts.user_defined, @@ -1794,99 +2462,132 @@ impl DefaultObjectUsecase { opts.clone(), ) .await; - if dsc.replicate_any() { + if replication.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(), + replication.pending_status().unwrap_or_default(), ); } - let mut reader = PutObjReader::new(hrd); - let cache_adapter = self.object_data_cache(); - let _ = invalidate_object_data_cache_before_mutation(&cache_adapter, &bucket, &fpath).await; + // Never wait while retaining a fully constructed replication + // decision. The lifecycle gate bounds builders, and this byte gate + // accounts every serial or staged member until storage returns. + let manager = get_concurrency_manager(); - let (obj_info, backfilled_old_current_size) = match store - .put_object_with_old_current_size(&bucket, &fpath, &mut reader, &opts) - .await - { - Ok(result) => result, - Err(e) => { - if should_ignore_extract_member_write_error(extract_options.ignore_errors, &member_read_failed) { - warn!(error = %e, "Archive object write skipped due to ignore-errors"); - continue; - } - return Err(ApiError::from(e).into()); - } - }; - let extract_versioned = BucketVersioningSys::prefix_enabled(&bucket, &fpath).await; - let post_commit_error = match quota_accounting_object_size(&obj_info, extract_quota_enabled) { - Ok(committed_size) => { - match previous_current_size_from_backfill(backfilled_old_current_size) { - Some(previous_current_size) => { - if extract_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; - } - } - None => { - record_bucket_object_write_unknown_previous_memory(&bucket, committed_size, extract_versioned).await; - } - } - None - } - Err(err) => Some(err), - }; - let _ = invalidate_object_data_cache_after_put_success(&cache_adapter, &bucket, &fpath).await; + let member_size = extract_try!( + usize::try_from(size) + .map_err(|_| object_s3_error(S3ErrorCode::InvalidArgument, "Snowball member size does not fit into usize")) + ); + let staging_weight = extract_try!(checked_extract_member_staging_weight(&fpath, member_size, &opts, &replication,)); + let action = batch_state.action(&fpath, member_size, staging_weight, max_inflight, durable_quota); - // Reuse the per-entry pre-commit decision (see `dsc` above) so the - // persisted pending marker and the schedule always agree. - if dsc.replicate_any() { - schedule_object_replication(obj_info.clone(), store.clone(), dsc).await; + if matches!(action, ExtractBatchAction::FlushThenStage | ExtractBatchAction::SerialBarrier) { + extract_try!( + flush_extract_batch(&mut batch, &mut batch_state, &commit_context, extract_options.ignore_errors,).await + ); } - if !wrote_any_entry { - rustfs_scanner::record_dirty_usage_bucket(&bucket); - wrote_any_entry = true; + let mut staging_permit = try_acquire_extract_staging_permit(manager, staging_weight); + if staging_permit.is_err() && !batch.is_empty() { + extract_try!( + flush_extract_batch(&mut batch, &mut batch_state, &commit_context, extract_options.ignore_errors,).await + ); + staging_permit = try_acquire_extract_staging_permit(manager, staging_weight); + } + let staging_permit = match staging_permit { + Ok(permit) => permit, + Err(error) if extract_options.ignore_errors => { + warn!(error = %error, "Archive object staging skipped due to ignore-errors"); + continue; + } + Err(error) => extract_try!(Err::(error)), + }; + + if action == ExtractBatchAction::SerialBarrier { + // Large and durable-quota members retain the legacy streaming + // path. Await the owner inline so the producer never advances + // the TAR while its Entry is alive. The complete commit and + // notification aggregation move into that owner before ECStore + // starts, so caller cancellation cannot interrupt a mutation + // after its quorum commit point. + let member_context = commit_context.clone(); + let owner_context = commit_context.clone(); + if is_dir { + drop(f); + extract_try!( + run_extract_outcomes_owner(owner_context, extract_options.ignore_errors, async move { + vec![ + commit_extract_member( + member_context, + ExtractPreparedMember { + archive_seq: extracted_entry_count, + key: fpath, + size, + actual_size, + body: std::io::Cursor::new(Vec::new()), + write_plan, + opts, + replication, + staging_permit, + member_permit, + }, + ) + .await, + ] + }) + .await + ); + } else { + extract_try!( + run_extract_outcomes_owner(owner_context, extract_options.ignore_errors, async move { + vec![ + commit_extract_member( + member_context, + ExtractPreparedMember { + archive_seq: extracted_entry_count, + key: fpath, + size, + actual_size, + body: f, + write_plan, + opts, + replication, + staging_permit, + member_permit, + }, + ) + .await, + ] + }) + .await + ); + } + continue; } - let _manager = get_concurrency_manager(); - let _fpath_clone = fpath.clone(); - let _bucket_clone = bucket.clone(); - let e_tag = obj_info.etag.clone().map(|etag| to_s3s_etag(&etag)); - - let output = PutObjectOutput { - e_tag, - ..Default::default() + let body = if is_dir { + drop(f); + ExtractStagedBody::empty() + } else { + extract_try!(stage_extract_member_body(&mut f, member_size).await) }; - - let actual_version_id = extract_notification_version_id(obj_info.version_id, opts.versioned, opts.version_suspended); - let mut event_object = convert_ecstore_object_info(obj_info.clone()); - event_object.version_id = (!actual_version_id.is_empty()).then_some(actual_version_id.clone()); - - let event_args = rustfs_notify::EventArgs { - event_name: put_event_name_for_post_object(false), - bucket_name: bucket.clone(), - object: event_object, - req_params: req_params.clone(), - resp_elements: build_event_resp_elements(&S3Response::new(output.clone()), &request_context.request_id), - version_id: actual_version_id, - host: host.clone(), - port, - user_agent: user_agent.clone(), - }; - - let notify = notify.clone(); - spawn_background_with_context(Some(request_context.clone()), async move { - notify.notify(event_args).await; + batch_state.record(&fpath, staging_weight); + batch.push(ExtractPreparedMember { + archive_seq: extracted_entry_count, + key: fpath, + size, + actual_size, + body, + write_plan, + opts, + replication, + staging_permit, + member_permit, }); - - if let Some(err) = post_commit_error { - return Err(err); - } } + extract_try!(flush_extract_batch(&mut batch, &mut batch_state, &commit_context, extract_options.ignore_errors,).await); let mut checksums = PutObjectChecksums { crc32: input.checksum_crc32, @@ -1937,6 +2638,7 @@ impl DefaultObjectUsecase { }; let result = Ok(S3Response::new(output)); let _ = helper.complete(&result); + put_request_guard.finish_ok(); result } } @@ -1949,6 +2651,524 @@ mod tests { use tokio::io::AsyncReadExt; use tokio_tar::{Builder, EntryType, Header}; + struct RecordingNotify { + events: tokio::sync::mpsc::UnboundedSender, + } + + #[async_trait::async_trait] + impl crate::runtime_sources::NotifyInterface for RecordingNotify { + async fn notify(&self, args: rustfs_notify::EventArgs) { + let _ = self.events.send(args.version_id); + } + + async fn add_event_specific_rules( + &self, + _bucket_name: &str, + _region: &str, + _event_rules: &[(Vec, String, String, Vec)], + ) -> Result<(), rustfs_notify::NotificationError> { + Ok(()) + } + + async fn clear_bucket_notification_rules(&self, _bucket_name: &str) -> Result<(), rustfs_notify::NotificationError> { + Ok(()) + } + } + + async fn recording_extract_commit_context( + notify: Arc, + ) -> Arc { + Arc::new(ExtractCommitContext { + store: crate::app::gating_test_env::shared_gating_ecstore().await, + cache_adapter: ObjectDataCacheAdapter::disabled_arc(), + notify, + bucket: "snowball-owner-test".to_string(), + quota_enabled: false, + request_context: request_context::RequestContext::from_headers(&HeaderMap::new()), + req_params: hashbrown::HashMap::new(), + host: String::new(), + port: 0, + user_agent: String::new(), + wrote_any_entry: AtomicBool::new(false), + }) + } + + #[test] + fn snowball_max_inflight_has_a_serial_compatibility_floor_and_bounded_ceiling() { + assert_eq!(EXTRACT_DEFAULT_MAX_INFLIGHT, 1); + assert_eq!(normalize_put_object_extract_max_inflight(0), 1); + assert_eq!(normalize_put_object_extract_max_inflight(1), 1); + assert_eq!(normalize_put_object_extract_max_inflight(usize::MAX), EXTRACT_BATCH_MAX_MEMBERS); + assert_eq!( + select_put_object_extract_max_inflight(EXTRACT_BATCH_MAX_MEMBERS, false, false), + 1, + "requests that stop on write errors must preserve serial member semantics" + ); + assert_eq!( + select_put_object_extract_max_inflight(EXTRACT_BATCH_MAX_MEMBERS, true, false), + EXTRACT_BATCH_MAX_MEMBERS, + "ignore-errors requests may use bounded parallel member commits" + ); + assert_eq!( + select_put_object_extract_max_inflight(EXTRACT_BATCH_MAX_MEMBERS, true, true), + 1, + "quota accounting can fail after storage commit and must remain serial" + ); + } + + #[test] + fn snowball_batch_state_flushes_on_duplicates_limits_and_serial_barriers() { + let mut state = ExtractBatchState::default(); + assert_eq!( + state.action("first", 4096, 8192, EXTRACT_BATCH_MAX_MEMBERS, false), + ExtractBatchAction::Stage + ); + state.record("first", 8192); + assert_eq!( + state.action("first", 4096, 8192, EXTRACT_BATCH_MAX_MEMBERS, false), + ExtractBatchAction::FlushThenStage + ); + assert_eq!( + state.action("large", EXTRACT_SMALL_MEMBER_MAX_BYTES + 1, 8192, EXTRACT_BATCH_MAX_MEMBERS, false,), + ExtractBatchAction::SerialBarrier + ); + assert_eq!( + state.action("quota", 4096, 8192, EXTRACT_BATCH_MAX_MEMBERS, true), + ExtractBatchAction::SerialBarrier + ); + assert_eq!(state.action("compat", 4096, 8192, 1, false), ExtractBatchAction::SerialBarrier); + assert_eq!( + state.action( + "exact-small-boundary", + EXTRACT_SMALL_MEMBER_MAX_BYTES, + 8192, + EXTRACT_BATCH_MAX_MEMBERS, + false, + ), + ExtractBatchAction::Stage + ); + assert_eq!( + state.action( + "oversized-context", + 4096, + EXTRACT_BATCH_MAX_STAGING_BYTES + 1, + EXTRACT_BATCH_MAX_MEMBERS, + false, + ), + ExtractBatchAction::SerialBarrier + ); + + state.clear(); + state.staging_bytes = EXTRACT_BATCH_MAX_STAGING_BYTES - 1024; + assert_eq!( + state.action("exact-memory", 4096, 1024, EXTRACT_BATCH_MAX_MEMBERS, false), + ExtractBatchAction::Stage + ); + assert_eq!( + state.action("memory", 4096, 2048, EXTRACT_BATCH_MAX_MEMBERS, false), + ExtractBatchAction::FlushThenStage + ); + + state.clear(); + for index in 0..EXTRACT_BATCH_MAX_MEMBERS { + state.record(&format!("key-{index}"), 1); + } + assert_eq!( + state.action("overflow", 1, 1, EXTRACT_BATCH_MAX_MEMBERS, false), + ExtractBatchAction::FlushThenStage + ); + } + + #[test] + fn snowball_staging_weight_includes_body_capacity_path_and_frozen_metadata() { + let mut opts = ObjectOptions::default(); + opts.user_defined.insert("x-amz-meta-one".to_string(), "value".to_string()); + opts.eval_metadata = Some(HashMap::from([("auth-view".to_string(), "retained".to_string())])); + let path = "prefix/object"; + let weight = checked_extract_member_staging_weight(path, 1, &opts, &ReplicateDecision::new()) + .expect("valid metadata must have a weight"); + assert_eq!( + weight, + EXTRACT_SMALL_MEMBER_MAX_BYTES + + EXTRACT_MEMBER_CONTEXT_OVERHEAD_BYTES + + (2 * path.len()) + + "x-amz-meta-one".len() + + "value".len() + + EXTRACT_METADATA_ENTRY_OVERHEAD_BYTES + + "auth-view".len() + + "retained".len() + + EXTRACT_METADATA_ENTRY_OVERHEAD_BYTES + ); + let empty_weight = checked_extract_member_staging_weight(path, 0, &ObjectOptions::default(), &ReplicateDecision::new()) + .expect("empty member must have a weight"); + assert_eq!(empty_weight, EXTRACT_MEMBER_CONTEXT_OVERHEAD_BYTES + (2 * path.len())); + } + + #[test] + fn snowball_staging_weight_accounts_replication_target_capacity() { + let path = "replicated/object"; + let opts = ObjectOptions::default(); + let baseline = checked_extract_member_staging_weight(path, 0, &opts, &ReplicateDecision::new()) + .expect("empty replication decision must have a weight"); + + let mut target_name = String::with_capacity(128); + target_name.push_str("target"); + let mut replication = ReplicateDecision::new(); + replication.targets_map.insert(target_name, Default::default()); + let map_allocation = + checked_extract_hash_map_allocation(&replication.targets_map).expect("target map allocation must fit"); + let retained_strings = { + let (name, target) = replication.targets_map.iter_mut().next().expect("target must exist"); + let mut arn = String::with_capacity(EXTRACT_BATCH_MAX_STAGING_BYTES); + arn.push('a'); + target.arn = arn; + let mut id = String::with_capacity(256); + id.push_str("id"); + target.id = id; + name.capacity() + target.arn.capacity() + target.id.capacity() + }; + let weight = checked_extract_member_staging_weight(path, 0, &opts, &replication) + .expect("replication decision must have a bounded weight"); + assert_eq!(weight, baseline + retained_strings + map_allocation); + assert_eq!( + ExtractBatchState::default().action("replicated/object", 0, weight, EXTRACT_BATCH_MAX_MEMBERS, false), + ExtractBatchAction::SerialBarrier, + "a retained target capacity above the micro-batch budget must remain serial" + ); + } + + #[test] + fn snowball_serial_members_consume_the_global_staging_budget() { + let manager = ConcurrencyManager::new(); + let serial_weight = EXTRACT_BATCH_MAX_STAGING_BYTES + 1; + assert_eq!( + ExtractBatchState::default().action("large-context", 0, serial_weight, EXTRACT_BATCH_MAX_MEMBERS, false), + ExtractBatchAction::SerialBarrier + ); + + let first = try_acquire_extract_staging_permit(&manager, serial_weight) + .expect("the first serial member must reserve its retained context"); + let error = try_acquire_extract_staging_permit(&manager, serial_weight) + .expect_err("a second serial member must not exceed the global staging budget"); + assert_eq!(error.code(), &S3ErrorCode::SlowDown); + + drop(first); + assert!( + try_acquire_extract_staging_permit(&manager, serial_weight).is_ok(), + "serial staging capacity must be reusable after storage releases it" + ); + } + + #[test] + fn snowball_staging_admission_rejects_without_retaining_waiters() { + let manager = ConcurrencyManager::new(); + let full_budget = manager + .try_acquire_snowball_staging_bytes( + u32::try_from(SNOWBALL_STAGING_BYTES_LIMIT).expect("the staging budget must fit into u32"), + ) + .expect("the exact global staging budget must be available"); + let error = + try_acquire_extract_staging_permit(&manager, 1).expect_err("a saturated staging budget must reject immediately"); + assert_eq!(error.code(), &S3ErrorCode::SlowDown); + drop(full_budget); + + let error = try_acquire_extract_staging_permit(&manager, SNOWBALL_STAGING_BYTES_LIMIT + 1) + .expect_err("a single retained context larger than the global budget must reject"); + assert_eq!(error.code(), &S3ErrorCode::SlowDown); + } + + #[tokio::test] + async fn snowball_staging_rejects_truncated_and_oversized_member_bodies() { + let mut truncated = std::io::Cursor::new(b"ab".to_vec()); + let error = stage_extract_member_body(&mut truncated, 3) + .await + .err() + .expect("a truncated TAR member must fail staging"); + assert_eq!(error.code(), &S3ErrorCode::IncompleteBody); + + let mut oversized = std::io::Cursor::new(b"abc".to_vec()); + let error = stage_extract_member_body(&mut oversized, 2) + .await + .err() + .expect("a TAR member longer than its declared size must fail staging"); + assert_eq!(error.code(), &S3ErrorCode::UnexpectedContent); + } + + #[tokio::test] + async fn snowball_staging_consumes_each_tar_entry_before_advancing() { + let mut builder = Builder::new(Vec::new()); + for (path, payload) in [("first", b"first-body".as_slice()), ("second", b"second-body".as_slice())] { + let mut header = Header::new_gnu(); + header.set_size(payload.len() as u64); + header.set_cksum(); + builder + .append_data(&mut header, path, std::io::Cursor::new(payload)) + .await + .expect("TAR member must append"); + } + let archive_bytes = builder.into_inner().await.expect("TAR fixture must finalize"); + let mut archive = Archive::new(std::io::Cursor::new(archive_bytes)); + let mut entries = archive.entries().expect("TAR entries must open"); + + let mut first = entries + .next() + .await + .expect("first entry must exist") + .expect("first entry must parse"); + let mut staged = stage_extract_member_body(&mut first, "first-body".len()) + .await + .expect("first body must stage"); + let mut staged_bytes = Vec::new(); + staged.read_to_end(&mut staged_bytes).await.expect("staged body must read"); + assert_eq!(staged_bytes, b"first-body"); + drop(first); + + let mut second = entries + .next() + .await + .expect("second entry must exist") + .expect("second entry must parse"); + let mut second_bytes = Vec::new(); + second.read_to_end(&mut second_bytes).await.expect("second body must read"); + assert_eq!(second_bytes, b"second-body"); + drop(second); + drop(entries); + assert!(archive.into_inner().is_ok(), "no TAR entry may escape the sequential producer"); + } + + #[test] + fn snowball_batch_error_selection_uses_archive_order() { + let outcomes = vec![ + ExtractCommitOutcome { + archive_seq: 2, + event: None, + error: Some(ExtractCommitError::Fatal(object_s3_error(S3ErrorCode::InvalidArgument, "later"))), + }, + ExtractCommitOutcome { + archive_seq: 1, + event: None, + error: Some(ExtractCommitError::Fatal(object_s3_error(S3ErrorCode::NoSuchKey, "earlier"))), + }, + ]; + let (_, error) = ordered_extract_outcomes(outcomes, false); + let error = error.expect("the earliest archive error must be returned"); + assert_eq!(error.code(), &S3ErrorCode::NoSuchKey); + } + + #[test] + fn snowball_batch_notifications_follow_archive_order() { + let outcomes = vec![ + ExtractCommitOutcome { + archive_seq: 2, + event: Some(EventArgsBuilder::default().version_id("second").build()), + error: None, + }, + ExtractCommitOutcome { + archive_seq: 1, + event: Some(EventArgsBuilder::default().version_id("first").build()), + error: None, + }, + ]; + let (events, error) = ordered_extract_outcomes(outcomes, false); + assert!(error.is_none(), "successful outcomes must be accepted"); + assert_eq!( + events.iter().map(|event| event.version_id.as_str()).collect::>(), + ["first", "second"] + ); + } + + #[tokio::test] + async fn snowball_notification_dispatch_detaches_slow_delivery() { + let started = Arc::new(tokio::sync::Notify::new()); + spawn_extract_notification_batch(None, vec![EventArgsBuilder::default().version_id("slow-target").build()], { + let started = started.clone(); + move |_event| { + let started = started.clone(); + async move { + started.notify_one(); + std::future::pending::<()>().await; + } + } + }); + + tokio::time::timeout(std::time::Duration::from_secs(1), started.notified()) + .await + .expect("detached notification delivery must start without blocking the caller"); + } + + #[tokio::test] + #[serial_test::serial] + async fn snowball_commit_owner_survives_request_cancellation() { + let entered = Arc::new(tokio::sync::Barrier::new(2)); + let release = Arc::new(tokio::sync::Barrier::new(2)); + let (events, mut received_events) = tokio::sync::mpsc::unbounded_channel(); + let context = recording_extract_commit_context(Arc::new(RecordingNotify { events })).await; + let manager = ConcurrencyManager::new(); + let owner_permit = manager + .acquire_snowball_member_commit() + .await + .expect("owner lifecycle admission must remain open"); + let mut other_permits = Vec::with_capacity(SNOWBALL_MEMBER_COMMIT_LIMIT - 1); + for _ in 1..SNOWBALL_MEMBER_COMMIT_LIMIT { + other_permits.push( + manager + .acquire_snowball_member_commit() + .await + .expect("remaining lifecycle admission must remain open"), + ); + } + + let request = spawn_traced_join({ + let entered = entered.clone(); + let release = release.clone(); + async move { + run_extract_outcomes_owner(context, false, async move { + let owner_permit = owner_permit; + entered.wait().await; + release.wait().await; + let outcome = ExtractCommitOutcome { + archive_seq: 1, + event: Some(EventArgsBuilder::default().version_id("cancelled-owner-event").build()), + error: None, + }; + drop(owner_permit); + vec![outcome] + }) + .await + } + }); + tokio::time::timeout(std::time::Duration::from_secs(1), entered.wait()) + .await + .expect("commit owner must start before the request waits for storage"); + request.abort(); + let join_error = request.await.expect_err("request task must be cancelled"); + assert!(join_error.is_cancelled()); + assert!( + manager.try_acquire_snowball_member_commit().is_none(), + "a cancelled request must not admit a 33rd member while its detached tail is running" + ); + release.wait().await; + let replacement = tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + if let Some(permit) = manager.try_acquire_snowball_member_commit() { + break permit; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("commit owner must finish after request cancellation"); + let event = tokio::time::timeout(std::time::Duration::from_secs(1), received_events.recv()) + .await + .expect("the detached owner must dispatch its committed event") + .expect("the notification recorder must remain open"); + assert_eq!(event, "cancelled-owner-event"); + assert!( + matches!( + received_events.try_recv(), + Err(tokio::sync::mpsc::error::TryRecvError::Empty | tokio::sync::mpsc::error::TryRecvError::Disconnected) + ), + "the detached owner must dispatch the committed event exactly once" + ); + drop((replacement, other_permits)); + } + + #[test] + fn snowball_ignore_errors_skips_storage_write_failure_and_keeps_later_success() { + let outcomes = vec![ + ExtractCommitOutcome { + archive_seq: 1, + event: None, + error: Some(ExtractCommitError::StorageWrite(object_s3_error( + S3ErrorCode::InternalError, + "injected write failure", + ))), + }, + ExtractCommitOutcome { + archive_seq: 2, + event: Some(EventArgsBuilder::default().version_id("later-success").build()), + error: None, + }, + ]; + + let (events, error) = ordered_extract_outcomes(outcomes, true); + assert!(error.is_none(), "ignore-errors must skip a storage-only write failure"); + assert_eq!(events.len(), 1); + assert_eq!(events[0].version_id, "later-success"); + + let slowdown = ExtractCommitError::StorageWrite(object_s3_error(S3ErrorCode::SlowDown, "injected admission rejection")); + assert!( + slowdown.into_unignored(true).is_none(), + "ignore-errors must treat member admission rejection as a skipped write" + ); + let slowdown = ExtractCommitError::StorageWrite(object_s3_error(S3ErrorCode::SlowDown, "injected admission rejection")); + assert_eq!( + slowdown + .into_unignored(false) + .expect("non-ignore requests must return admission rejection") + .code(), + &S3ErrorCode::SlowDown + ); + } + + #[test] + fn snowball_ignore_errors_never_skips_reader_or_other_fatal_failures() { + let outcomes = vec![ExtractCommitOutcome { + archive_seq: 1, + event: None, + error: Some(ExtractCommitError::Fatal(object_s3_error( + S3ErrorCode::IncompleteBody, + "injected reader failure", + ))), + }]; + + let (_, error) = ordered_extract_outcomes(outcomes, true); + let error = error.expect("ignore-errors must not hide reader, codec, length, resource, or post-commit failures"); + assert_eq!(error.code(), &S3ErrorCode::IncompleteBody); + } + + #[tokio::test] + async fn snowball_batch_runner_polls_commits_concurrently_and_drains_every_outcome() { + let completed = Arc::new(AtomicUsize::new(0)); + let active = Arc::new(AtomicUsize::new(0)); + let max_active = Arc::new(AtomicUsize::new(0)); + let barrier = Arc::new(tokio::sync::Barrier::new(3)); + let outcomes = tokio::time::timeout( + std::time::Duration::from_secs(1), + run_extract_commits(1..=3, { + let active = active.clone(); + let max_active = max_active.clone(); + let barrier = barrier.clone(); + let completed = completed.clone(); + move |archive_seq| { + let active = active.clone(); + let max_active = max_active.clone(); + let barrier = barrier.clone(); + let completed = completed.clone(); + async move { + let current = active.fetch_add(1, Ordering::AcqRel) + 1; + max_active.fetch_max(current, Ordering::AcqRel); + barrier.wait().await; + active.fetch_sub(1, Ordering::AcqRel); + completed.fetch_add(1, Ordering::Relaxed); + ExtractCommitOutcome { + archive_seq, + event: None, + error: (archive_seq == 2) + .then(|| ExtractCommitError::Fatal(object_s3_error(S3ErrorCode::InvalidArgument, "injected"))), + } + } + } + }), + ) + .await + .expect("the production batch runner must poll all commits concurrently"); + assert_eq!(outcomes.len(), 3); + assert_eq!(completed.load(Ordering::Relaxed), 3); + assert_eq!(max_active.load(Ordering::Acquire), 3); + assert_eq!(outcomes.iter().filter(|outcome| outcome.error.is_some()).count(), 1); + } + #[test] fn archive_format_uses_only_the_ambiguous_zlib_extension_as_a_fallback() { assert_eq!( @@ -3294,11 +4514,18 @@ mod tests { assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); assert!(failed.load(Ordering::Acquire)); - assert!(!should_ignore_extract_member_write_error(true, &failed)); + let classified = classify_extract_member_write_error(object_s3_error_default(S3ErrorCode::IncompleteBody), &failed) + .into_unignored(true) + .expect("a reader-backed store error must remain fatal"); + assert_eq!(classified.code(), &S3ErrorCode::IncompleteBody); let storage_only_failure = AtomicBool::new(false); - assert!(should_ignore_extract_member_write_error(true, &storage_only_failure)); - assert!(!should_ignore_extract_member_write_error(false, &storage_only_failure)); + assert!( + classify_extract_member_write_error(object_s3_error_default(S3ErrorCode::InternalError), &storage_only_failure) + .into_unignored(true) + .is_none(), + "a storage-only write failure may be ignored" + ); } #[tokio::test] diff --git a/rustfs/src/app/object/put.rs b/rustfs/src/app/object/put.rs index 73398cd74..673498e9b 100644 --- a/rustfs/src/app/object/put.rs +++ b/rustfs/src/app/object/put.rs @@ -308,14 +308,14 @@ pub(crate) fn guard_put_object_body_read_timeout( }) } -struct PooledBufferReader { +pub(super) struct PooledBufferReader { buffer: PooledBuffer, len: usize, pos: usize, } impl PooledBufferReader { - fn new(buffer: PooledBuffer, len: usize) -> Self { + pub(super) fn new(buffer: PooledBuffer, len: usize) -> Self { Self { buffer, len, pos: 0 } } } @@ -631,7 +631,7 @@ fn select_put_path_with_concurrency( /// where the allocation cost is negligible (≤4KiB memcpy). const POOL_BYPASS_MAX_SIZE: usize = 4 * 1024; -async fn read_small_put_body_into(body: &mut R, buf: &mut B, size: usize) -> S3Result<()> +pub(super) async fn read_small_put_body_into(body: &mut R, buf: &mut B, size: usize) -> S3Result<()> where R: AsyncRead + Unpin, B: bytes::BufMut, diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 38b372dbc..8eb19d5d4 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -978,9 +978,12 @@ pub(crate) mod bucket { } pub(crate) mod concurrency { + #[cfg(test)] + pub(crate) use crate::storage::storage_api::concurrency_consumer::SNOWBALL_MEMBER_COMMIT_LIMIT; pub(crate) use crate::storage::storage_api::concurrency_consumer::{ ConcurrencyManager, DiskReadAdmission, ForegroundWriteAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, - PutObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size, + PutObjectGuard, SNOWBALL_STAGING_BYTES_LIMIT, get_concurrency_aware_buffer_size, get_concurrency_manager, + get_put_concurrency_aware_buffer_size, }; } diff --git a/rustfs/src/storage/concurrency/manager.rs b/rustfs/src/storage/concurrency/manager.rs index 4c5d20ef0..779f9b5c0 100644 --- a/rustfs/src/storage/concurrency/manager.rs +++ b/rustfs/src/storage/concurrency/manager.rs @@ -31,10 +31,12 @@ use rustfs_io_metrics::bandwidth::{BandwidthMonitor, BandwidthSnapshot}; use rustfs_io_metrics::{MetricsCollector, PerformanceMetrics}; use std::sync::{Arc, LazyLock, Mutex}; use std::time::Duration; -use tokio::sync::Semaphore; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use tracing::debug; const DERIVED_LARGE_PUT_ADMISSION_LIMIT_MAX: usize = 32; +pub(crate) const SNOWBALL_MEMBER_COMMIT_LIMIT: usize = 32; +pub(crate) const SNOWBALL_STAGING_BYTES_LIMIT: usize = 4 * MI_B; /// Global concurrency manager instance pub(crate) static CONCURRENCY_MANAGER: LazyLock = LazyLock::new(ConcurrencyManager::new); @@ -69,6 +71,12 @@ pub struct ConcurrencyManager { metrics_collector: Arc, /// Foreground write admission policy, resolved once at startup. foreground_write_admission_policy: ForegroundWriteAdmissionPolicy, + /// Snowball members are internal PUTs, so they use a separate global gate + /// from preparation through the independently owned post-commit tail. + snowball_member_commit_semaphore: Arc, + /// Bounds the owned member bodies and metadata retained between TAR parsing + /// and storage commit across all extract requests. + snowball_staging_bytes_semaphore: Arc, } impl std::fmt::Debug for ConcurrencyManager { @@ -417,6 +425,8 @@ impl ConcurrencyManager { bandwidth_monitor, metrics_collector, foreground_write_admission_policy, + snowball_member_commit_semaphore: Arc::new(Semaphore::new(SNOWBALL_MEMBER_COMMIT_LIMIT)), + snowball_staging_bytes_semaphore: Arc::new(Semaphore::new(SNOWBALL_STAGING_BYTES_LIMIT)), } } @@ -556,6 +566,40 @@ impl ConcurrencyManager { .await } + /// Admit a Snowball member through the foreground PUT policy using the + /// member's logical size. The outer archive has a separate preflight, while + /// every member shares the ordinary PUT gate and its wait/rejection policy. + pub(crate) async fn admit_snowball_foreground_write( + &self, + member_size: i64, + ) -> Result { + self.foreground_write_admission_policy + .admit(ForegroundWriteAdmissionKind::PutObject, member_size) + .await + } + + /// Acquire one global Snowball member lifecycle slot. + pub(crate) async fn acquire_snowball_member_commit(&self) -> Result { + self.snowball_member_commit_semaphore.clone().acquire_owned().await + } + + /// Try to acquire one global Snowball member lifecycle slot. + pub(crate) fn try_acquire_snowball_member_commit(&self) -> Option { + self.snowball_member_commit_semaphore.clone().try_acquire_owned().ok() + } + + /// Try to reserve prepared-member bytes without waiting. + /// + /// A producer holding a non-empty micro-batch must use this method and + /// flush before waiting, otherwise several archives can each retain part of + /// the global budget while waiting forever for the remainder. + pub(crate) fn try_acquire_snowball_staging_bytes(&self, bytes: u32) -> Option { + self.snowball_staging_bytes_semaphore + .clone() + .try_acquire_many_owned(bytes) + .ok() + } + /// Admit a multipart UploadPart request under the configured write gate. /// /// Multipart workloads can saturate memory and internode write streams with @@ -1050,13 +1094,138 @@ impl Default for ConcurrencyManager { mod integration_tests { use super::super::io_schedule::{IoLoadLevel, IoPriority}; use super::super::request_guard::GetObjectGuard; - use super::{ConcurrencyManager, ForegroundWriteAdmission, derive_large_put_admission_limit}; + use super::{ + ConcurrencyManager, ForegroundWriteAdmission, SNOWBALL_MEMBER_COMMIT_LIMIT, SNOWBALL_STAGING_BYTES_LIMIT, + derive_large_put_admission_limit, + }; use crate::storage::storage_api::concurrency_consumer::PutObjectGuard; use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass}; use rustfs_io_core::io_profile::{AccessPattern, StorageMedia}; use serial_test::serial; use std::time::Duration; + #[test] + fn test_snowball_gates_are_global_bounded_and_reusable() { + let manager = ConcurrencyManager::new(); + let clone = manager.clone(); + + let commit_permits = manager + .snowball_member_commit_semaphore + .clone() + .try_acquire_many_owned(u32::try_from(SNOWBALL_MEMBER_COMMIT_LIMIT).expect("Snowball commit limit must fit into u32")) + .expect("the exact Snowball commit limit must be available"); + assert!( + clone.snowball_member_commit_semaphore.clone().try_acquire_owned().is_err(), + "a cloned manager must share the global commit gate" + ); + drop(commit_permits); + assert!(clone.snowball_member_commit_semaphore.clone().try_acquire_owned().is_ok()); + + let staging_bytes = u32::try_from(SNOWBALL_STAGING_BYTES_LIMIT).expect("Snowball staging limit must fit into u32"); + let staging_permit = manager + .try_acquire_snowball_staging_bytes(staging_bytes) + .expect("the exact Snowball staging budget must be available"); + assert!( + clone.try_acquire_snowball_staging_bytes(1).is_none(), + "a cloned manager must share the global staging budget" + ); + drop(staging_permit); + assert!(clone.try_acquire_snowball_staging_bytes(1).is_some()); + } + + #[tokio::test] + async fn test_snowball_members_share_the_strict_foreground_put_gate() { + let manager = ConcurrencyManager::with_put_admission_for_test(true, 2, Duration::ZERO); + let outer = match manager + .admit_put_object(1) + .await + .expect("strict outer admission must remain open") + { + ForegroundWriteAdmission::Admitted(permit) => permit, + outcome => panic!("strict outer admission must return a permit: {outcome:?}"), + }; + let member = match manager + .admit_snowball_foreground_write(1) + .await + .expect("strict member admission must remain open") + { + ForegroundWriteAdmission::Admitted(permit) => permit, + outcome => panic!("strict member admission must return a permit: {outcome:?}"), + }; + assert!( + matches!( + manager + .admit_snowball_foreground_write(1) + .await + .expect("strict member admission must remain open"), + ForegroundWriteAdmission::Rejected + ), + "a saturated Snowball member admission must preserve the zero-wait rejection policy" + ); + assert!( + matches!( + manager.admit_put_object(1).await.expect("strict gate must remain usable"), + ForegroundWriteAdmission::Rejected + ), + "outer PUTs and Snowball members must exhaust the same strict gate" + ); + + drop(outer); + let replacement = match manager + .admit_snowball_foreground_write(1) + .await + .expect("released strict capacity must be reusable") + { + ForegroundWriteAdmission::Admitted(permit) => permit, + outcome => panic!("strict replacement admission must return a permit: {outcome:?}"), + }; + drop((member, replacement)); + } + + #[tokio::test] + async fn test_snowball_members_use_their_size_for_the_large_foreground_put_gate() { + let min_size = 16 * 1024 * 1024; + let manager = ConcurrencyManager::with_large_put_admission_for_test(true, 1, min_size, Duration::ZERO); + + assert!(matches!( + manager + .admit_put_object((min_size - 1) as i64) + .await + .expect("small outer archive admission must remain open"), + ForegroundWriteAdmission::Disabled + )); + let large_member = match manager + .admit_snowball_foreground_write(min_size as i64) + .await + .expect("large Snowball member admission must remain open") + { + ForegroundWriteAdmission::Admitted(permit) => permit, + outcome => panic!("large Snowball member must consume the large PUT gate: {outcome:?}"), + }; + assert!(matches!( + manager + .admit_snowball_foreground_write(min_size as i64) + .await + .expect("saturated Snowball member admission must remain open"), + ForegroundWriteAdmission::Rejected + )); + assert!(matches!( + manager + .admit_put_object(min_size as i64) + .await + .expect("ordinary large PUT admission must remain open"), + ForegroundWriteAdmission::Rejected + )); + assert!(matches!( + manager + .admit_snowball_foreground_write((min_size - 1) as i64) + .await + .expect("small Snowball member admission must remain open"), + ForegroundWriteAdmission::Disabled + )); + drop(large_member); + } + #[tokio::test] #[serial] async fn test_concurrency_manager_priority_queue_integration() { diff --git a/rustfs/src/storage/concurrency/mod.rs b/rustfs/src/storage/concurrency/mod.rs index 7996649ab..03f52abb9 100644 --- a/rustfs/src/storage/concurrency/mod.rs +++ b/rustfs/src/storage/concurrency/mod.rs @@ -51,6 +51,9 @@ pub use io_schedule::{ pub use request_guard::{GetObjectGuard, PutObjectGuard}; // Concurrency manager +#[cfg(test)] +pub(crate) use manager::SNOWBALL_MEMBER_COMMIT_LIMIT; +pub(crate) use manager::SNOWBALL_STAGING_BYTES_LIMIT; pub use manager::{ConcurrencyManager, DiskReadAdmission, ForegroundWriteAdmission}; // ============================================ diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 41cceca63..f82adf0ce 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -125,9 +125,12 @@ pub(crate) mod access_consumer { } pub(crate) mod concurrency_consumer { + #[cfg(test)] + pub(crate) use super::super::concurrency::SNOWBALL_MEMBER_COMMIT_LIMIT; pub(crate) use super::super::concurrency::{ ConcurrencyManager, DiskReadAdmission, ForegroundWriteAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, - PutObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size, + PutObjectGuard, SNOWBALL_STAGING_BYTES_LIMIT, get_concurrency_aware_buffer_size, get_concurrency_manager, + get_put_concurrency_aware_buffer_size, }; } diff --git a/rustfs/src/table_catalog/tests.rs b/rustfs/src/table_catalog/tests.rs index 40f3d7c88..6ed547c03 100644 --- a/rustfs/src/table_catalog/tests.rs +++ b/rustfs/src/table_catalog/tests.rs @@ -16445,7 +16445,7 @@ fn object_mutation_entrypoints_call_reserved_prefix_guard() { "if let Err(err) = validate_table_catalog_object_mutation(&bucket, &obj_id.key).await", "validate_table_catalog_object_mutation(&bucket, &object).await?;", "validate_object_key(&key, \"PUT\")?;\n validate_table_catalog_object_mutation(&bucket, &key).await?;", - "validate_table_catalog_object_mutation(&bucket, &fpath).await?;", + "extract_try!(validate_table_catalog_object_mutation(&bucket, &fpath).await);", ] { assert!(source.contains(expected), "missing object mutation guard: {expected}"); }