diff --git a/crates/ecstore/src/bucket/quota/reservation.rs b/crates/ecstore/src/bucket/quota/reservation.rs index 15ec725cd..175785487 100644 --- a/crates/ecstore/src/bucket/quota/reservation.rs +++ b/crates/ecstore/src/bucket/quota/reservation.rs @@ -624,9 +624,11 @@ async fn settle(data: &LedgerReservationData, committed: bool) -> Result<()> { let ledger_object = data.ledger_object.clone(); let operation_id = data.operation_id; let reservation = data.reservation.clone(); - let bucket = data.bucket.clone(); tokio::spawn(async move { - fence_namespace_mutations(&store, &bucket, &reservation.object, reservation.target()).await?; + // The commit/abort path releases its own object fence before settlement. + // Do not revoke all tokens here: a deferred retry can run after the + // object lock is released and would otherwise revoke a later write's + // newly acquired fence for the same object. let ledger_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &ledger_object).await?; let ledger_guard = ledger_lock.get_write_lock(get_lock_acquire_timeout()).await?; fence_namespace_mutations(&store, RUSTFS_META_BUCKET, &ledger_object, None).await?; diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index 13a45fd35..1f3ac8856 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -2901,7 +2901,7 @@ async fn handle_misc_extension_request(req: &mut S3Request, route: &MiscEx MiscExtRoute::ObjectLambda { bucket, object } => { let get_req = build_object_lambda_get_request(req, bucket, object)?; let usecase = default_object_usecase(); - let get_resp = Box::pin(usecase.execute_get_object(get_req)).await?; + let get_resp = usecase.execute_get_object(get_req).await?; invoke_object_lambda_target(req, bucket, object, get_resp).await } MiscExtRoute::ListenNotification { bucket } => { diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index aefea0046..74fd41a18 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -556,9 +556,13 @@ pub(super) fn apply_quota_admission(opts: &mut ObjectOptions, result: &QuotaChec "Bucket quota check temporarily unavailable, please retry".to_string(), )); }; - if current_usage <= quota_limit { - let _ = opts.set_quota_admission(current_usage, quota_limit); + if current_usage > quota_limit { + return Err(S3Error::with_message( + S3ErrorCode::InvalidRequest, + format!("Bucket quota exceeded. Current usage: {current_usage} bytes, limit: {quota_limit} bytes"), + )); } + let _ = opts.set_quota_admission(current_usage, quota_limit); Ok(()) } @@ -6540,15 +6544,12 @@ impl DefaultObjectUsecase { }) } - pub fn execute_get_object( - &self, - req: S3Request, - ) -> std::pin::Pin>> + Send + '_>> { - Box::pin(self.execute_get_object_inner(req)) - } - #[instrument(level = "trace", skip(self, req))] #[hotpath::measure(impl_type = "DefaultObjectUsecase")] + pub async fn execute_get_object(&self, req: S3Request) -> S3Result> { + Box::pin(self.execute_get_object_inner(req)).await + } + async fn execute_get_object_inner(&self, req: S3Request) -> S3Result> { if let Some(context) = &self.context { let _ = context.object_store(); @@ -7003,14 +7004,11 @@ impl DefaultObjectUsecase { result } - pub fn execute_copy_object( - &self, - req: S3Request, - ) -> std::pin::Pin>> + Send + '_>> { - Box::pin(self.execute_copy_object_inner(req)) + #[instrument(level = "debug", skip(self, req))] + pub async fn execute_copy_object(&self, req: S3Request) -> S3Result> { + Box::pin(self.execute_copy_object_inner(req)).await } - #[instrument(level = "debug", skip(self, req))] async fn execute_copy_object_inner(&self, req: S3Request) -> S3Result> { if let Some(context) = &self.context { let _ = context.object_store(); @@ -18166,6 +18164,22 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); } + #[test] + fn legacy_quota_admission_rejects_already_over_limit() { + let result = QuotaCheckResult { + allowed: true, + current_usage: Some(6), + quota_limit: Some(5), + operation_size: 0, + remaining: Some(0), + uses_durable_reservations: false, + }; + let mut opts = ObjectOptions::default(); + let err = + apply_quota_admission(&mut opts, &result).expect_err("legacy completion must not bypass an already exceeded quota"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } + #[test] fn quota_admission_fails_closed_on_checker_error() { // A configured hard quota must never be bypassed by an internal fault: a checker error becomes a retryable ServiceUnavailable, not a silent allow. diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index e27c6dc29..891897f31 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -301,7 +301,7 @@ impl S3 for FS { #[instrument(level = "debug", skip(self, req))] async fn copy_object(&self, req: S3Request) -> S3Result> { let usecase = s3_api::object_usecase_for(self); - Box::pin(usecase.execute_copy_object(req)).await + usecase.execute_copy_object(req).await } #[instrument( @@ -704,7 +704,7 @@ impl S3 for FS { async fn get_object(&self, req: S3Request) -> S3Result> { crate::hp_guard!("S3::get_object"); let usecase = s3_api::object_usecase_for(self); - Box::pin(usecase.execute_get_object(req)).await + usecase.execute_get_object(req).await } async fn get_object_acl(&self, req: S3Request) -> S3Result> {