gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-15 09:33:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(quota): close legacy and deferred settlement bypasses

Browse Source
This commit is contained in:
马登山
2026-08-13 17:35:52 +08:00
parent be84adf8a9
commit 2610801136
4 changed files with 36 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/quota/reservation.rs
+4 -2
View File
@@ -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?;
rustfs/src/admin/router.rs
+1 -1
View File
@@ -2901,7 +2901,7 @@ async fn handle_misc_extension_request(req: &mut S3Request<Body>, 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 } => {
rustfs/src/app/object_usecase.rs
+29 -15
View File
@@ -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<GetObjectInput>,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = S3Result<S3Response<GetObjectOutput>>> + 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<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
Box::pin(self.execute_get_object_inner(req)).await
}
async fn execute_get_object_inner(&self, req: S3Request<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
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<CopyObjectInput>,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = S3Result<S3Response<CopyObjectOutput>>> + Send + '_>> {
Box::pin(self.execute_copy_object_inner(req))
#[instrument(level = "debug", skip(self, req))]
pub async fn execute_copy_object(&self, req: S3Request<CopyObjectInput>) -> S3Result<S3Response<CopyObjectOutput>> {
Box::pin(self.execute_copy_object_inner(req)).await
}
#[instrument(level = "debug", skip(self, req))]
async fn execute_copy_object_inner(&self, req: S3Request<CopyObjectInput>) -> S3Result<S3Response<CopyObjectOutput>> {
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.
rustfs/src/storage/ecfs.rs
+2 -2
View File
@@ -301,7 +301,7 @@ impl S3 for FS {
#[instrument(level = "debug", skip(self, req))]
async fn copy_object(&self, req: S3Request<CopyObjectInput>) -> S3Result<S3Response<CopyObjectOutput>> {
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<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
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<GetObjectAclInput>) -> S3Result<S3Response<GetObjectAclOutput>> {