mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-27 06:18:43 +00:00
fix(quota): keep ledger-admitted writes independent of scanner usage (#8699)
Hard quota admission is owned by the durable reservation ledger (decision D3, rustfs/backlog#2872), but the request-path quota check still read scanner usage on every write and mapped `UsageUnavailable` to a retryable 503 even for buckets the ledger admits. For a durable quota the checker now treats scanner usage as reporting input: an unavailable lookup yields `current_usage = None` and `allowed = true`, counted and logged at debug level, and admin quota stats report the usage as null instead of zero. Buckets without a reservation protocol keep the fail-closed snapshot path. Quotas saved before the reservation protocols are re-saved under protocol v2 with the same limit, type and created_at. The write goes through a new guarded save that keeps the admin save's lifecycle/transaction guards, incarnation check and fleet capability proof and additionally refuses when the quota revision moved since the cached read, so a concurrent quota edit wins. It is triggered lazily from the quota read path (background, one attempt per bucket with a retry interval, never blocking the read) and in bulk by `POST /rustfs/admin/v3/quota-migration`, which reports every bucket. The ADR's Consequences section inventories the remaining scanner-usage "authority" consumers with their meaning after D3, retirement condition and compatibility window; the operations guide documents the migration and the report-only usage semantics. Refs rustfs/backlog#2904, rustfs/backlog#2872, rustfs/rustfs#8678, rustfs/rustfs#5716, rustfs/rustfs#7350 Co-authored-by: Hauser <housemecn@gmail.com>
This commit is contained in:
Chris
@@ -252,11 +252,11 @@ pub mod bucket {
|
||||
pub mod quota {
|
||||
pub use crate::bucket::quota::{
|
||||
BucketQuota, QUOTA_RESERVATION_PROTOCOL_V1, QUOTA_RESERVATION_PROTOCOL_V2, QuotaCheckResult, QuotaError,
|
||||
QuotaOperation,
|
||||
QuotaMigrationSkipReason, QuotaOperation, QuotaProtocolMigration,
|
||||
};
|
||||
|
||||
pub mod checker {
|
||||
pub use crate::bucket::quota::checker::QuotaChecker;
|
||||
pub use crate::bucket::quota::checker::{QuotaChecker, record_reservation_protocol_migration};
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1027,6 +1027,52 @@ async fn update_quota_if_incarnation_stamped(
|
||||
update_under_config_write_guard(sys, &guard, rustfs_config::QUOTA_CONFIG_FILE, data, updated_at).await
|
||||
}
|
||||
|
||||
/// Rewrite the quota configuration only while the bucket still carries
|
||||
/// `expected_incarnation_id` and `expected_quota_revision` (its
|
||||
/// `quota_config_updated_at`), under the same guards and fleet proof as
|
||||
/// [`update_quota_if_incarnation`].
|
||||
///
|
||||
/// The revision check turns a stale read into a typed
|
||||
/// [`Error::PreconditionFailed`] instead of a lost update: the reservation
|
||||
/// protocol migration (rustfs/backlog#2904) rewrites a configuration it read
|
||||
/// from the cache, and an admin save landing in between must win. The system
|
||||
/// is passed explicitly so the write publishes into the cache the caller
|
||||
/// reads from.
|
||||
pub async fn update_quota_with_sys_if_incarnation_and_revision(
|
||||
sys: Arc<RwLock<BucketMetadataSys>>,
|
||||
bucket: &str,
|
||||
data: Vec<u8>,
|
||||
expected_incarnation_id: Uuid,
|
||||
expected_quota_revision: OffsetDateTime,
|
||||
proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken,
|
||||
) -> Result<OffsetDateTime> {
|
||||
let guard = Box::pin(acquire_config_write_guard_for_incarnation(
|
||||
sys.clone(),
|
||||
bucket,
|
||||
Some(expected_incarnation_id),
|
||||
))
|
||||
.await?;
|
||||
if !crate::services::notification_sys::cross_pool_fence_fleet_proof_matches(proof) {
|
||||
return Err(Error::NamespaceLockQuorumUnavailable {
|
||||
mode: "quota_capability",
|
||||
bucket: bucket.to_string(),
|
||||
object: rustfs_config::QUOTA_CONFIG_FILE.to_string(),
|
||||
required: 1,
|
||||
achieved: 0,
|
||||
});
|
||||
}
|
||||
let metadata_sys = sys.read().await.clone();
|
||||
match Box::pin(metadata_sys.read_authoritative_metadata_from_disk_under_transaction_lock(bucket)).await? {
|
||||
BucketMetadataAuthority::Authoritative(current) if current.quota_config_updated_at == expected_quota_revision => {}
|
||||
BucketMetadataAuthority::Authoritative(_) => return Err(Error::PreconditionFailed),
|
||||
BucketMetadataAuthority::MissingBucket => return Err(Error::BucketNotFound(bucket.to_string())),
|
||||
BucketMetadataAuthority::Fabricated => {
|
||||
return Err(Error::other("bucket quota metadata is not authoritative"));
|
||||
}
|
||||
}
|
||||
update_under_config_write_guard(sys, &guard, rustfs_config::QUOTA_CONFIG_FILE, data, None).await
|
||||
}
|
||||
|
||||
pub async fn update_bucket_targets_under_transaction_lock(
|
||||
guard: &BucketMetadataMutationGuard,
|
||||
bucket: &str,
|
||||
|
||||
@@ -12,16 +12,33 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::{BucketQuota, QuotaCheckResult, QuotaError, QuotaOperation};
|
||||
use crate::bucket::metadata_sys::{BucketMetadataSys, update, update_if_incarnation};
|
||||
use super::{
|
||||
BucketQuota, QUOTA_RESERVATION_PROTOCOL_V2, QuotaCheckResult, QuotaError, QuotaMigrationSkipReason, QuotaOperation,
|
||||
QuotaProtocolMigration,
|
||||
};
|
||||
use crate::bucket::metadata_sys::{
|
||||
BucketMetadataSys, update, update_if_incarnation, update_quota_with_sys_if_incarnation_and_revision,
|
||||
};
|
||||
use crate::data_usage::get_bucket_usage_memory;
|
||||
use crate::error::StorageError;
|
||||
use rustfs_config::QUOTA_CONFIG_FILE;
|
||||
use rustfs_scanner_metrics::metrics::Metric;
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, Mutex, OnceLock, PoisonError};
|
||||
use std::time::{Duration, Instant};
|
||||
use time::OffsetDateTime;
|
||||
use tokio::sync::RwLock;
|
||||
use tracing::{debug, warn};
|
||||
use tracing::{debug, info, warn};
|
||||
|
||||
const LOG_COMPONENT_ECSTORE: &str = "ecstore";
|
||||
const LOG_SUBSYSTEM_QUOTA: &str = "quota";
|
||||
const EVENT_QUOTA_USAGE_REPORT: &str = "quota_usage_report";
|
||||
const EVENT_QUOTA_PROTOCOL_MIGRATION: &str = "quota_reservation_protocol_migration";
|
||||
const METRIC_QUOTA_USAGE_REPORT_UNAVAILABLE_TOTAL: &str = "rustfs_quota_usage_report_unavailable_total";
|
||||
const METRIC_QUOTA_PROTOCOL_MIGRATIONS_TOTAL: &str = "rustfs_quota_reservation_protocol_migrations_total";
|
||||
/// A lazy migration attempt that did not settle the bucket is retried no
|
||||
/// sooner than this; the admin command is not rate limited.
|
||||
const LAZY_MIGRATION_RETRY_INTERVAL: Duration = Duration::from_secs(300);
|
||||
|
||||
pub struct QuotaChecker {
|
||||
metadata_sys: Arc<RwLock<BucketMetadataSys>>,
|
||||
@@ -74,7 +91,15 @@ impl QuotaChecker {
|
||||
Some(q) => q,
|
||||
};
|
||||
|
||||
let current_usage = self.get_real_time_usage(bucket).await?;
|
||||
// A durable quota is admitted by the reservation ledger at commit, so
|
||||
// scanner usage is reporting input here: an unavailable lookup is
|
||||
// reported as absent instead of failing the write (rustfs/backlog#2904).
|
||||
// The snapshot protocol still admits from this value and fails closed.
|
||||
let current_usage = if uses_durable_reservations {
|
||||
self.report_only_usage(bucket).await
|
||||
} else {
|
||||
Some(self.get_real_time_usage(bucket).await?)
|
||||
};
|
||||
|
||||
// The reporting path projects this operation; storage mutations reserve it at commit.
|
||||
let admission_size = if uses_durable_reservations && !force_usage_calculation {
|
||||
@@ -82,27 +107,23 @@ impl QuotaChecker {
|
||||
} else {
|
||||
operation_size
|
||||
};
|
||||
let expected_usage = match operation {
|
||||
let expected_usage = current_usage.map(|current_usage| match operation {
|
||||
QuotaOperation::PutObject | QuotaOperation::PostObject | QuotaOperation::CopyObject => {
|
||||
current_usage.saturating_add(admission_size)
|
||||
}
|
||||
QuotaOperation::DeleteObject => current_usage.saturating_sub(operation_size),
|
||||
};
|
||||
});
|
||||
|
||||
let allowed = match operation {
|
||||
QuotaOperation::PutObject | QuotaOperation::PostObject | QuotaOperation::CopyObject => {
|
||||
quota_config.check_operation_allowed(current_usage, admission_size)
|
||||
current_usage.is_none_or(|current_usage| quota_config.check_operation_allowed(current_usage, admission_size))
|
||||
}
|
||||
QuotaOperation::DeleteObject => true,
|
||||
};
|
||||
|
||||
let remaining = if quota_limit >= expected_usage {
|
||||
Some(quota_limit - expected_usage)
|
||||
} else {
|
||||
Some(0)
|
||||
};
|
||||
let remaining = expected_usage.map(|expected_usage| quota_limit.saturating_sub(expected_usage));
|
||||
|
||||
if !allowed {
|
||||
if let (false, Some(current_usage)) = (allowed, current_usage) {
|
||||
warn!(
|
||||
"Quota exceeded for bucket: {}, current: {}, limit: {}, attempted: {}",
|
||||
bucket, current_usage, quota_limit, operation_size
|
||||
@@ -111,7 +132,7 @@ impl QuotaChecker {
|
||||
|
||||
let result = QuotaCheckResult {
|
||||
allowed,
|
||||
current_usage: Some(current_usage),
|
||||
current_usage,
|
||||
quota_limit: Some(quota_limit),
|
||||
operation_size,
|
||||
remaining,
|
||||
@@ -152,6 +173,10 @@ impl QuotaChecker {
|
||||
reason: format!("Failed to parse quota config: {}", e),
|
||||
})?;
|
||||
|
||||
if quota.needs_reservation_protocol_migration() {
|
||||
self.schedule_reservation_protocol_migration(bucket).await;
|
||||
}
|
||||
|
||||
Ok(quota)
|
||||
}
|
||||
|
||||
@@ -219,9 +244,146 @@ impl QuotaChecker {
|
||||
}
|
||||
|
||||
let quota = self.get_quota_config(bucket).await?;
|
||||
let current_usage = self.get_real_time_usage(bucket).await?;
|
||||
let current_usage = if quota.uses_durable_reservations() {
|
||||
self.report_only_usage(bucket).await
|
||||
} else {
|
||||
Some(self.get_real_time_usage(bucket).await?)
|
||||
};
|
||||
|
||||
Ok((quota, Some(current_usage)))
|
||||
Ok((quota, current_usage))
|
||||
}
|
||||
|
||||
/// Scanner usage for a bucket the reservation ledger admits is reporting
|
||||
/// input only, so a failed lookup degrades to "unknown" and is counted
|
||||
/// instead of becoming an error the write path would fail on.
|
||||
async fn report_only_usage(&self, bucket: &str) -> Option<u64> {
|
||||
match self.get_real_time_usage(bucket).await {
|
||||
Ok(usage) => Some(usage),
|
||||
Err(err) => {
|
||||
metrics::counter!(METRIC_QUOTA_USAGE_REPORT_UNAVAILABLE_TOTAL).increment(1);
|
||||
debug!(
|
||||
event = EVENT_QUOTA_USAGE_REPORT,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_QUOTA,
|
||||
state = "usage_unknown",
|
||||
bucket = %bucket,
|
||||
error = %err,
|
||||
"durable quota usage report degraded"
|
||||
);
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Re-save `bucket`'s quota under the sharded durable reservation
|
||||
/// protocol (rustfs/backlog#2904). The limit and type are unchanged. The
|
||||
/// write takes the incarnation- and revision-checked quota save path with
|
||||
/// the fleet capability proof, so a mixed-version fleet fails the attempt
|
||||
/// and a concurrent quota edit or bucket replacement leaves the newer
|
||||
/// configuration untouched.
|
||||
pub async fn migrate_reservation_protocol(&self, bucket: &str) -> Result<QuotaProtocolMigration, QuotaError> {
|
||||
let (meta, _) = self
|
||||
.metadata_sys
|
||||
.read()
|
||||
.await
|
||||
.get_config(bucket)
|
||||
.await
|
||||
.map_err(QuotaError::StorageError)?;
|
||||
if meta.quota_config_json.is_empty() {
|
||||
return Ok(QuotaProtocolMigration::NoQuota);
|
||||
}
|
||||
let quota: BucketQuota = serde_json::from_slice(&meta.quota_config_json).map_err(|e| QuotaError::InvalidConfig {
|
||||
reason: format!("Failed to parse quota config: {}", e),
|
||||
})?;
|
||||
if let Some(reservation_protocol) = quota.reservation_protocol {
|
||||
return Ok(if quota.uses_durable_reservations() {
|
||||
QuotaProtocolMigration::AlreadyDurable { reservation_protocol }
|
||||
} else {
|
||||
QuotaProtocolMigration::Skipped {
|
||||
reason: QuotaMigrationSkipReason::UnsupportedReservationProtocol,
|
||||
}
|
||||
});
|
||||
}
|
||||
let Some(migrated) = quota.migrated_to_sharded_reservations() else {
|
||||
return Ok(QuotaProtocolMigration::NoQuota);
|
||||
};
|
||||
if !meta.bucket_incarnation_sidecar || meta.bucket_incarnation_id.is_nil() {
|
||||
return Err(QuotaError::StorageError(StorageError::other(
|
||||
"bucket incarnation metadata is not authoritative",
|
||||
)));
|
||||
}
|
||||
// Same gate as the admin save: a durable protocol must not reach a
|
||||
// fleet that still has nodes that would fail closed on it.
|
||||
let proof = crate::services::notification_sys::acquire_cross_pool_fence_fleet_proof().ok_or_else(|| {
|
||||
QuotaError::StorageError(StorageError::NamespaceLockQuorumUnavailable {
|
||||
mode: "quota_capability",
|
||||
bucket: bucket.to_string(),
|
||||
object: QUOTA_CONFIG_FILE.to_string(),
|
||||
required: 1,
|
||||
achieved: 0,
|
||||
})
|
||||
})?;
|
||||
let json_data = serde_json::to_vec(&migrated).map_err(|e| QuotaError::InvalidConfig {
|
||||
reason: format!("Failed to serialize quota config: {}", e),
|
||||
})?;
|
||||
let start_time = Instant::now();
|
||||
let saved = update_quota_with_sys_if_incarnation_and_revision(
|
||||
Arc::clone(&self.metadata_sys),
|
||||
bucket,
|
||||
json_data,
|
||||
meta.bucket_incarnation_id,
|
||||
meta.quota_config_updated_at,
|
||||
&proof,
|
||||
)
|
||||
.await;
|
||||
match saved {
|
||||
Ok(updated_at) => {
|
||||
rustfs_scanner_metrics::metrics::Metrics::inc_time(Metric::QuotaSync, start_time.elapsed());
|
||||
Ok(QuotaProtocolMigration::Migrated {
|
||||
reservation_protocol: QUOTA_RESERVATION_PROTOCOL_V2,
|
||||
updated_at,
|
||||
})
|
||||
}
|
||||
Err(StorageError::PreconditionFailed) => Ok(QuotaProtocolMigration::Skipped {
|
||||
reason: QuotaMigrationSkipReason::QuotaRevisionChanged,
|
||||
}),
|
||||
Err(StorageError::BucketNotFound(_)) => Ok(QuotaProtocolMigration::Skipped {
|
||||
reason: QuotaMigrationSkipReason::BucketIncarnationChanged,
|
||||
}),
|
||||
Err(err) => Err(QuotaError::StorageError(err)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Lazy half of the migration: the first quota read of a bucket still on
|
||||
/// the snapshot protocol re-saves it under the sharded durable protocol
|
||||
/// in the background; the read itself is neither blocked nor failed by
|
||||
/// it. Only the metadata system registered on the store's instance
|
||||
/// context migrates, so the rewrite publishes into the cache every
|
||||
/// reader of this store consults; a checker over an ad-hoc system
|
||||
/// (tooling, tests) only reads. An attempt that does not settle the
|
||||
/// bucket is retried no sooner than `LAZY_MIGRATION_RETRY_INTERVAL`.
|
||||
async fn schedule_reservation_protocol_migration(&self, bucket: &str) {
|
||||
let registered = self.metadata_sys.read().await.object_store().ctx.bucket_metadata_sys();
|
||||
if !registered.is_some_and(|registered| Arc::ptr_eq(®istered, &self.metadata_sys)) {
|
||||
return;
|
||||
}
|
||||
if !lazy_migration_attempt_due(bucket) {
|
||||
return;
|
||||
}
|
||||
let checker = Self::new(Arc::clone(&self.metadata_sys));
|
||||
let bucket = bucket.to_string();
|
||||
tokio::spawn(async move {
|
||||
let outcome = checker.migrate_reservation_protocol(&bucket).await;
|
||||
record_reservation_protocol_migration(&bucket, "lazy", &outcome);
|
||||
if matches!(
|
||||
outcome,
|
||||
Ok(QuotaProtocolMigration::Migrated { .. }
|
||||
| QuotaProtocolMigration::AlreadyDurable { .. }
|
||||
| QuotaProtocolMigration::NoQuota)
|
||||
) {
|
||||
lazy_migration_attempt_settled(&bucket);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
pub async fn bucket_exists(&self, bucket: &str) -> bool {
|
||||
@@ -262,10 +424,107 @@ impl QuotaChecker {
|
||||
}
|
||||
}
|
||||
|
||||
/// One entry per bucket whose lazy migration is in flight or did not settle;
|
||||
/// the entry is dropped once the bucket settles, so the map is bounded by the
|
||||
/// number of buckets still on the snapshot protocol.
|
||||
fn lazy_migration_attempts() -> &'static Mutex<HashMap<String, Instant>> {
|
||||
static ATTEMPTS: OnceLock<Mutex<HashMap<String, Instant>>> = OnceLock::new();
|
||||
ATTEMPTS.get_or_init(Mutex::default)
|
||||
}
|
||||
|
||||
fn lazy_migration_attempt_due(bucket: &str) -> bool {
|
||||
let mut attempts = lazy_migration_attempts().lock().unwrap_or_else(PoisonError::into_inner);
|
||||
let now = Instant::now();
|
||||
if attempts
|
||||
.get(bucket)
|
||||
.is_some_and(|last| now.duration_since(*last) < LAZY_MIGRATION_RETRY_INTERVAL)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
attempts.insert(bucket.to_string(), now);
|
||||
true
|
||||
}
|
||||
|
||||
fn lazy_migration_attempt_settled(bucket: &str) {
|
||||
lazy_migration_attempts()
|
||||
.lock()
|
||||
.unwrap_or_else(PoisonError::into_inner)
|
||||
.remove(bucket);
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn lazy_migration_attempt_pending(bucket: &str) -> bool {
|
||||
lazy_migration_attempts()
|
||||
.lock()
|
||||
.unwrap_or_else(PoisonError::into_inner)
|
||||
.contains_key(bucket)
|
||||
}
|
||||
|
||||
/// One log line and one counter per migration attempt; `trigger` names the
|
||||
/// caller (`lazy` from the quota read path, `admin` from the admin command).
|
||||
pub fn record_reservation_protocol_migration(
|
||||
bucket: &str,
|
||||
trigger: &'static str,
|
||||
outcome: &Result<QuotaProtocolMigration, QuotaError>,
|
||||
) {
|
||||
let result = match outcome {
|
||||
Ok(QuotaProtocolMigration::Migrated { .. }) => "migrated",
|
||||
Ok(QuotaProtocolMigration::AlreadyDurable { .. }) | Ok(QuotaProtocolMigration::NoQuota) => "noop",
|
||||
Ok(QuotaProtocolMigration::Skipped { .. }) => "skipped",
|
||||
Err(_) => "failed",
|
||||
};
|
||||
metrics::counter!(METRIC_QUOTA_PROTOCOL_MIGRATIONS_TOTAL, "trigger" => trigger, "result" => result).increment(1);
|
||||
match outcome {
|
||||
Ok(QuotaProtocolMigration::Migrated {
|
||||
reservation_protocol, ..
|
||||
}) => info!(
|
||||
event = EVENT_QUOTA_PROTOCOL_MIGRATION,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_QUOTA,
|
||||
result,
|
||||
trigger,
|
||||
bucket = %bucket,
|
||||
reservation_protocol,
|
||||
"quota reservation protocol migrated"
|
||||
),
|
||||
Ok(QuotaProtocolMigration::Skipped { reason }) => debug!(
|
||||
event = EVENT_QUOTA_PROTOCOL_MIGRATION,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_QUOTA,
|
||||
result,
|
||||
trigger,
|
||||
bucket = %bucket,
|
||||
reason = reason.as_str(),
|
||||
"quota reservation protocol migration skipped"
|
||||
),
|
||||
Ok(_) => debug!(
|
||||
event = EVENT_QUOTA_PROTOCOL_MIGRATION,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_QUOTA,
|
||||
result,
|
||||
trigger,
|
||||
bucket = %bucket,
|
||||
"quota reservation protocol migration not needed"
|
||||
),
|
||||
Err(err) => warn!(
|
||||
event = EVENT_QUOTA_PROTOCOL_MIGRATION,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_QUOTA,
|
||||
result,
|
||||
trigger,
|
||||
bucket = %bucket,
|
||||
error = %err,
|
||||
"quota reservation protocol migration failed"
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::bucket::metadata_sys::test_support::isolated_store_over_temp_disks;
|
||||
use crate::bucket::quota::QuotaType;
|
||||
use crate::error::StorageError;
|
||||
use serial_test::serial;
|
||||
use uuid::Uuid;
|
||||
|
||||
@@ -378,6 +637,367 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
async fn isolated_durable_quota_store() -> (Vec<tempfile::TempDir>, Arc<crate::store::ECStore>, Arc<RwLock<BucketMetadataSys>>)
|
||||
{
|
||||
let (dirs, store) = isolated_store_over_temp_disks().await;
|
||||
crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
|
||||
let sys = crate::bucket::metadata_sys::bucket_metadata_sys_of(&store.ctx).expect("scoped metadata system");
|
||||
(dirs, store, sys)
|
||||
}
|
||||
|
||||
async fn create_bucket_with_quota_json(store: &Arc<crate::store::ECStore>, bucket: &str, quota_json: Vec<u8>) {
|
||||
use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
|
||||
store
|
||||
.make_bucket(bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("create quota test bucket");
|
||||
store
|
||||
.update_bucket_metadata_config(bucket, rustfs_config::QUOTA_CONFIG_FILE, quota_json)
|
||||
.await
|
||||
.expect("persist quota config");
|
||||
}
|
||||
|
||||
/// Regression (rustfs/backlog#2904, decision D3 in rustfs/backlog#2872): a
|
||||
/// bucket whose quota is admitted by the durable reservation ledger must
|
||||
/// not fail its write because scanner usage is unavailable. The checker
|
||||
/// reports `current_usage = None` instead of `UsageUnavailable`, while the
|
||||
/// ledger still rejects an over-limit write and a bucket without a
|
||||
/// reservation protocol keeps failing closed.
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn durable_quota_check_reports_unknown_usage_without_failing_the_write() {
|
||||
use crate::object_api::{ObjectOptions, PutObjReader};
|
||||
use crate::storage_api_contracts::object::ObjectIO as _;
|
||||
|
||||
let (_dirs, store, sys) = isolated_durable_quota_store().await;
|
||||
let checker = QuotaChecker::new(Arc::clone(&sys));
|
||||
let bucket = format!("quota-durable-{}", Uuid::new_v4().simple());
|
||||
create_bucket_with_quota_json(
|
||||
&store,
|
||||
&bucket,
|
||||
serde_json::to_vec(&BucketQuota::new_sharded(4096)).expect("sharded quota should encode"),
|
||||
)
|
||||
.await;
|
||||
crate::data_usage::invalidate_data_usage_snapshot_cache().await;
|
||||
|
||||
let result = checker
|
||||
.check_quota(&bucket, QuotaOperation::PutObject, 512)
|
||||
.await
|
||||
.expect("a ledger-admitted bucket must not fail the check without scanner usage");
|
||||
assert!(result.allowed);
|
||||
assert!(result.uses_durable_reservations);
|
||||
assert_eq!(result.quota_limit, Some(4096));
|
||||
assert_eq!(result.current_usage, None, "unknown usage is reported as absent, never as zero");
|
||||
assert_eq!(result.remaining, None);
|
||||
|
||||
let (_, usage) = checker
|
||||
.get_quota_stats(&bucket)
|
||||
.await
|
||||
.expect("stats for a ledger-admitted bucket degrade to an absent usage");
|
||||
assert_eq!(usage, None);
|
||||
|
||||
// Negative control: the ledger, not the scanner, still refuses an
|
||||
// over-limit write with the same error.
|
||||
let mut too_large = PutObjReader::from_vec(vec![0x5a; 8192]);
|
||||
let err = store
|
||||
.put_object(&bucket, "too-large", &mut too_large, &ObjectOptions::default())
|
||||
.await
|
||||
.expect_err("the ledger must reject a write above the quota limit");
|
||||
assert!(matches!(err, StorageError::QuotaExceeded { .. }), "{err:?}");
|
||||
|
||||
let mut fits = PutObjReader::from_vec(vec![0x5a; 512]);
|
||||
store
|
||||
.put_object(&bucket, "fits", &mut fits, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("a write within the quota limit commits through the ledger");
|
||||
|
||||
// A bucket without a reservation protocol keeps the fail-closed path.
|
||||
let legacy = format!("quota-legacy-{}", Uuid::new_v4().simple());
|
||||
create_bucket_with_quota_json(&store, &legacy, br#"{"quota":4096,"quota_type":"Hard"}"#.to_vec()).await;
|
||||
assert!(matches!(
|
||||
checker.check_quota(&legacy, QuotaOperation::PutObject, 512).await,
|
||||
Err(QuotaError::UsageUnavailable { .. })
|
||||
));
|
||||
assert!(matches!(checker.get_quota_stats(&legacy).await, Err(QuotaError::UsageUnavailable { .. })));
|
||||
|
||||
crate::data_usage::prepare_bucket_usage_for_namespace_change(&bucket, None)
|
||||
.await
|
||||
.expect("test usage cache cleanup should succeed");
|
||||
crate::data_usage::invalidate_data_usage_snapshot_cache().await;
|
||||
}
|
||||
|
||||
fn legacy_quota_json(limit: u64) -> Vec<u8> {
|
||||
format!(r#"{{"quota":{limit},"quota_type":"Hard","created_at":"2024-01-01T00:00:00Z"}}"#).into_bytes()
|
||||
}
|
||||
|
||||
async fn persisted_quota(store: &Arc<crate::store::ECStore>, bucket: &str) -> BucketQuota {
|
||||
let (quota, _, _) = crate::bucket::metadata_sys::get_quota_config_and_incarnation_from_disk_in(&store.ctx, bucket)
|
||||
.await
|
||||
.expect("quota metadata should load from disk");
|
||||
quota.expect("a quota should be persisted")
|
||||
}
|
||||
|
||||
/// rustfs/backlog#2904: a quota saved before the reservation protocols is
|
||||
/// re-saved under the sharded durable protocol with the same limit, type
|
||||
/// and audit timestamp; durable and unknown protocols are left alone.
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn legacy_quota_migrates_to_sharded_reservation_protocol_once() {
|
||||
let (_dirs, store, sys) = isolated_durable_quota_store().await;
|
||||
crate::services::notification_sys::install_cross_pool_fence_fleet_proof_for_test();
|
||||
let checker = QuotaChecker::new(Arc::clone(&sys));
|
||||
let bucket = format!("quota-migrate-{}", Uuid::new_v4().simple());
|
||||
create_bucket_with_quota_json(&store, &bucket, legacy_quota_json(4096)).await;
|
||||
|
||||
let outcome = checker
|
||||
.migrate_reservation_protocol(&bucket)
|
||||
.await
|
||||
.expect("migration should succeed");
|
||||
assert!(
|
||||
matches!(
|
||||
outcome,
|
||||
QuotaProtocolMigration::Migrated {
|
||||
reservation_protocol: QUOTA_RESERVATION_PROTOCOL_V2,
|
||||
..
|
||||
}
|
||||
),
|
||||
"{outcome:?}"
|
||||
);
|
||||
let migrated = persisted_quota(&store, &bucket).await;
|
||||
assert!(migrated.uses_sharded_reservations());
|
||||
assert_eq!(migrated.quota, Some(4096));
|
||||
assert_eq!(migrated.quota_type, QuotaType::Hard);
|
||||
assert_eq!(migrated.created_at.map(|at| at.unix_timestamp()), Some(1_704_067_200));
|
||||
assert_eq!(
|
||||
checker.get_quota_config(&bucket).await.expect("cached quota should load"),
|
||||
migrated,
|
||||
"the rewrite publishes into the registered cache"
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
checker
|
||||
.migrate_reservation_protocol(&bucket)
|
||||
.await
|
||||
.expect("second attempt should succeed"),
|
||||
QuotaProtocolMigration::AlreadyDurable {
|
||||
reservation_protocol: QUOTA_RESERVATION_PROTOCOL_V2
|
||||
}
|
||||
);
|
||||
|
||||
let no_quota = format!("quota-none-{}", Uuid::new_v4().simple());
|
||||
create_bucket_with_quota_json(
|
||||
&store,
|
||||
&no_quota,
|
||||
serde_json::to_vec(&BucketQuota::new(None)).expect("empty quota should encode"),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
checker
|
||||
.migrate_reservation_protocol(&no_quota)
|
||||
.await
|
||||
.expect("no quota is a no-op"),
|
||||
QuotaProtocolMigration::NoQuota
|
||||
);
|
||||
|
||||
let unsupported = format!("quota-unsupported-{}", Uuid::new_v4().simple());
|
||||
let unsupported_json = br#"{"quota":0,"quota_type":"Hard","reservation_protocol":3,"reservation_quota":2048}"#.to_vec();
|
||||
create_bucket_with_quota_json(&store, &unsupported, unsupported_json.clone()).await;
|
||||
assert_eq!(
|
||||
checker
|
||||
.migrate_reservation_protocol(&unsupported)
|
||||
.await
|
||||
.expect("an unknown protocol is skipped, never rewritten"),
|
||||
QuotaProtocolMigration::Skipped {
|
||||
reason: QuotaMigrationSkipReason::UnsupportedReservationProtocol
|
||||
}
|
||||
);
|
||||
let (raw, _) = sys.read().await.get_config(&unsupported).await.expect("cached metadata");
|
||||
assert_eq!(raw.quota_config_json, unsupported_json);
|
||||
}
|
||||
|
||||
/// The guarded save refuses a replaced bucket and a quota edited since the
|
||||
/// cached read; the migration reports the latter as a skip and leaves the
|
||||
/// newer configuration for the next attempt.
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn quota_protocol_migration_requires_matching_incarnation_and_revision() {
|
||||
let (_dirs, store, sys) = isolated_durable_quota_store().await;
|
||||
crate::services::notification_sys::install_cross_pool_fence_fleet_proof_for_test();
|
||||
let checker = QuotaChecker::new(Arc::clone(&sys));
|
||||
let bucket = format!("quota-cas-{}", Uuid::new_v4().simple());
|
||||
create_bucket_with_quota_json(&store, &bucket, legacy_quota_json(4096)).await;
|
||||
let (meta, _) = sys.read().await.get_config(&bucket).await.expect("cached metadata");
|
||||
let proof = crate::services::notification_sys::acquire_cross_pool_fence_fleet_proof().expect("fleet proof is installed");
|
||||
let data = serde_json::to_vec(&BucketQuota::new_sharded(4096)).expect("sharded quota should encode");
|
||||
|
||||
let err = update_quota_with_sys_if_incarnation_and_revision(
|
||||
Arc::clone(&sys),
|
||||
&bucket,
|
||||
data.clone(),
|
||||
Uuid::new_v4(),
|
||||
meta.quota_config_updated_at,
|
||||
&proof,
|
||||
)
|
||||
.await
|
||||
.expect_err("a different incarnation must not be rewritten");
|
||||
assert!(matches!(err, StorageError::BucketNotFound(_)), "{err:?}");
|
||||
|
||||
let err = update_quota_with_sys_if_incarnation_and_revision(
|
||||
Arc::clone(&sys),
|
||||
&bucket,
|
||||
data,
|
||||
meta.bucket_incarnation_id,
|
||||
meta.quota_config_updated_at + time::Duration::seconds(1),
|
||||
&proof,
|
||||
)
|
||||
.await
|
||||
.expect_err("a different quota revision must not be rewritten");
|
||||
assert!(matches!(err, StorageError::PreconditionFailed), "{err:?}");
|
||||
assert_eq!(persisted_quota(&store, &bucket).await.quota, Some(4096));
|
||||
assert!(persisted_quota(&store, &bucket).await.needs_reservation_protocol_migration());
|
||||
|
||||
// Model a peer whose cache missed a newer save: the migration reads
|
||||
// the stale entry, the guarded write sees the newer revision, and the
|
||||
// newer configuration stays in place.
|
||||
let (stale, _) = sys.read().await.get_config(&bucket).await.expect("cached metadata");
|
||||
store
|
||||
.update_bucket_metadata_config(&bucket, QUOTA_CONFIG_FILE, legacy_quota_json(8192))
|
||||
.await
|
||||
.expect("newer quota save");
|
||||
sys.read().await.set(bucket.clone(), stale).await;
|
||||
assert_eq!(
|
||||
checker
|
||||
.migrate_reservation_protocol(&bucket)
|
||||
.await
|
||||
.expect("a stale read is a skip, not a failure"),
|
||||
QuotaProtocolMigration::Skipped {
|
||||
reason: QuotaMigrationSkipReason::QuotaRevisionChanged
|
||||
}
|
||||
);
|
||||
let on_disk = persisted_quota(&store, &bucket).await;
|
||||
assert_eq!(on_disk.quota, Some(8192));
|
||||
assert!(
|
||||
on_disk.needs_reservation_protocol_migration(),
|
||||
"the newer save is left for the next attempt"
|
||||
);
|
||||
}
|
||||
|
||||
/// The first quota read through the instance's registered metadata
|
||||
/// system migrates the bucket in the background and still answers with
|
||||
/// the configuration it found; a checker over an ad-hoc system only reads.
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn lazy_quota_protocol_migration_runs_from_the_quota_read_path() {
|
||||
let (_dirs, store, sys) = isolated_durable_quota_store().await;
|
||||
crate::services::notification_sys::install_cross_pool_fence_fleet_proof_for_test();
|
||||
let checker = QuotaChecker::new(Arc::clone(&sys));
|
||||
let bucket = format!("quota-lazy-{}", Uuid::new_v4().simple());
|
||||
create_bucket_with_quota_json(&store, &bucket, legacy_quota_json(4096)).await;
|
||||
|
||||
let first = checker.get_quota_config(&bucket).await.expect("legacy quota should load");
|
||||
assert!(
|
||||
first.needs_reservation_protocol_migration(),
|
||||
"the triggering read answers with what it found"
|
||||
);
|
||||
let deadline = Instant::now() + Duration::from_secs(30);
|
||||
loop {
|
||||
let on_disk = persisted_quota(&store, &bucket).await;
|
||||
if on_disk.uses_sharded_reservations() {
|
||||
assert_eq!(on_disk.quota, Some(4096));
|
||||
break;
|
||||
}
|
||||
assert!(Instant::now() < deadline, "lazy migration did not land");
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
}
|
||||
let settled = Instant::now() + Duration::from_secs(10);
|
||||
while lazy_migration_attempt_pending(&bucket) {
|
||||
assert!(Instant::now() < settled, "a migrated bucket must leave the retry map");
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
}
|
||||
|
||||
let ad_hoc = QuotaChecker::new(Arc::new(RwLock::new(BucketMetadataSys::new(Arc::clone(&store)))));
|
||||
let other = format!("quota-adhoc-{}", Uuid::new_v4().simple());
|
||||
create_bucket_with_quota_json(&store, &other, legacy_quota_json(4096)).await;
|
||||
assert!(
|
||||
ad_hoc
|
||||
.get_quota_config(&other)
|
||||
.await
|
||||
.expect("legacy quota should load")
|
||||
.needs_reservation_protocol_migration()
|
||||
);
|
||||
assert!(
|
||||
!lazy_migration_attempt_pending(&other),
|
||||
"an ad-hoc metadata system never schedules a migration"
|
||||
);
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
assert!(persisted_quota(&store, &other).await.needs_reservation_protocol_migration());
|
||||
}
|
||||
|
||||
/// Migration while writes are in flight: objects committed under the
|
||||
/// snapshot protocol before and around the rewrite are accounted exactly
|
||||
/// once by the sharded ledger's exact bootstrap, so the limit holds to
|
||||
/// the byte afterwards. (Under `cfg(test)` a handler-supplied snapshot
|
||||
/// admission short-circuits `begin`, so every concurrent write here takes
|
||||
/// the snapshot path; the bootstrap accounting is what is verified.)
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
||||
#[serial]
|
||||
async fn concurrent_puts_across_protocol_migration_are_accounted_exactly_once() {
|
||||
use crate::object_api::{ObjectOptions, PutObjReader};
|
||||
use crate::storage_api_contracts::object::ObjectIO as _;
|
||||
|
||||
const OBJECTS: u64 = 8;
|
||||
const SIZE: u64 = 256;
|
||||
const SLACK: u64 = 64;
|
||||
const LIMIT: u64 = OBJECTS * SIZE + SLACK;
|
||||
|
||||
let (_dirs, store, sys) = isolated_durable_quota_store().await;
|
||||
crate::services::notification_sys::install_cross_pool_fence_fleet_proof_for_test();
|
||||
let bucket = format!("quota-race-{}", Uuid::new_v4().simple());
|
||||
create_bucket_with_quota_json(&store, &bucket, legacy_quota_json(LIMIT)).await;
|
||||
|
||||
let mut writers = Vec::with_capacity(OBJECTS as usize);
|
||||
for index in 0..OBJECTS {
|
||||
let store = Arc::clone(&store);
|
||||
let bucket = bucket.clone();
|
||||
writers.push(tokio::spawn(async move {
|
||||
// The admission a request handler supplies on the snapshot
|
||||
// protocol; ignored once the ledger owns admission.
|
||||
let mut opts = ObjectOptions::default();
|
||||
assert!(opts.set_quota_admission(0, LIMIT));
|
||||
let mut body = PutObjReader::from_vec(vec![0x5a; SIZE as usize]);
|
||||
store.put_object(&bucket, &format!("object-{index}"), &mut body, &opts).await
|
||||
}));
|
||||
}
|
||||
let migration = {
|
||||
let checker = QuotaChecker::new(Arc::clone(&sys));
|
||||
let bucket = bucket.clone();
|
||||
tokio::spawn(async move { checker.migrate_reservation_protocol(&bucket).await })
|
||||
};
|
||||
for writer in writers {
|
||||
writer
|
||||
.await
|
||||
.expect("writer task should not panic")
|
||||
.expect("every write fits the limit");
|
||||
}
|
||||
let outcome = migration
|
||||
.await
|
||||
.expect("migration task should not panic")
|
||||
.expect("migration should succeed");
|
||||
assert!(matches!(outcome, QuotaProtocolMigration::Migrated { .. }), "{outcome:?}");
|
||||
|
||||
let mut over = PutObjReader::from_vec(vec![0x5a; (SLACK + 1) as usize]);
|
||||
let err = store
|
||||
.put_object(&bucket, "over", &mut over, &ObjectOptions::default())
|
||||
.await
|
||||
.expect_err("one byte above the remaining quota must be rejected");
|
||||
assert!(matches!(err, StorageError::QuotaExceeded { .. }), "{err:?}");
|
||||
let mut exact = PutObjReader::from_vec(vec![0x5a; SLACK as usize]);
|
||||
store
|
||||
.put_object(&bucket, "exact", &mut exact, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("exactly the remaining bytes fit");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_quota_check_no_limit() {
|
||||
let result = QuotaCheckResult {
|
||||
|
||||
@@ -164,6 +164,26 @@ impl BucketQuota {
|
||||
.is_some_and(|version| !matches!(version, QUOTA_RESERVATION_PROTOCOL_V1 | QUOTA_RESERVATION_PROTOCOL_V2))
|
||||
}
|
||||
|
||||
/// A limit saved before the reservation protocols existed: its writes
|
||||
/// are still admitted from scanner usage snapshots. Durable and unknown
|
||||
/// future protocols are never rewritten.
|
||||
pub fn needs_reservation_protocol_migration(&self) -> bool {
|
||||
self.quota.is_some() && self.reservation_protocol.is_none()
|
||||
}
|
||||
|
||||
/// The same limit and type re-expressed under the sharded durable
|
||||
/// reservation protocol; `created_at` is kept for audit continuity.
|
||||
/// `None` when the quota is not a migration candidate.
|
||||
pub fn migrated_to_sharded_reservations(&self) -> Option<Self> {
|
||||
self.needs_reservation_protocol_migration().then(|| Self {
|
||||
quota: self.quota,
|
||||
quota_type: self.quota_type.clone(),
|
||||
reservation_protocol: Some(QUOTA_RESERVATION_PROTOCOL_V2),
|
||||
created_at: self.created_at,
|
||||
updated_at: self.updated_at,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn check_operation_allowed(&self, current_usage: u64, operation_size: u64) -> bool {
|
||||
if operation_size == 0 {
|
||||
return true;
|
||||
@@ -200,6 +220,45 @@ pub enum QuotaOperation {
|
||||
DeleteObject,
|
||||
}
|
||||
|
||||
/// Outcome of re-saving one bucket's quota under the sharded durable
|
||||
/// reservation protocol (rustfs/backlog#2904).
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum QuotaProtocolMigration {
|
||||
/// No quota limit is configured; nothing to migrate.
|
||||
NoQuota,
|
||||
/// The quota already uses a durable reservation protocol.
|
||||
AlreadyDurable { reservation_protocol: u32 },
|
||||
/// The quota was re-saved under the sharded protocol at this revision.
|
||||
Migrated {
|
||||
reservation_protocol: u32,
|
||||
updated_at: OffsetDateTime,
|
||||
},
|
||||
/// The quota was left untouched for the stated reason.
|
||||
Skipped { reason: QuotaMigrationSkipReason },
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum QuotaMigrationSkipReason {
|
||||
/// A reservation protocol this node does not understand; rewriting it
|
||||
/// would downgrade a newer fleet's quota.
|
||||
UnsupportedReservationProtocol,
|
||||
/// The quota configuration changed between the cached read and the
|
||||
/// guarded write; the newer configuration wins.
|
||||
QuotaRevisionChanged,
|
||||
/// The bucket was replaced or removed since the cached read.
|
||||
BucketIncarnationChanged,
|
||||
}
|
||||
|
||||
impl QuotaMigrationSkipReason {
|
||||
pub fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::UnsupportedReservationProtocol => "unsupported_reservation_protocol",
|
||||
Self::QuotaRevisionChanged => "quota_revision_changed",
|
||||
Self::BucketIncarnationChanged => "bucket_incarnation_changed",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum QuotaError {
|
||||
#[error("Bucket quota exceeded: current={current}, limit={limit}, operation={operation}")]
|
||||
|
||||
@@ -49,7 +49,16 @@ lifecycle discovery, and heal discovery for everyone.
|
||||
- Buckets with `reservation_protocol = None` are migrated by re-saving their
|
||||
quota configuration through the existing save path; until migrated they keep
|
||||
the degraded floor path and fail closed only when neither authoritative usage
|
||||
nor a floor exists.
|
||||
nor a floor exists. The re-save is triggered lazily by the first quota read
|
||||
of such a bucket (`QuotaChecker::get_quota_config`) and in bulk by
|
||||
`POST /rustfs/admin/v3/quota-migration`; the limit and type are unchanged,
|
||||
the write keeps the incarnation, quota-revision and fleet-proof checks of
|
||||
the admin save, and a bucket whose quota changed in between is skipped,
|
||||
not overwritten. See
|
||||
[durable-quota-reservation-v2.md](../operations/durable-quota-reservation-v2.md).
|
||||
Note: the admin default for a new quota is still protocol v1
|
||||
(`BucketQuota::new`); v2 is chosen per request or by this migration. Both
|
||||
are durable, so neither is a migration candidate.
|
||||
- [scanner-per-bucket-convergence.md](scanner-per-bucket-convergence.md) no
|
||||
longer cites quota authority as a reason to keep uncertified buckets
|
||||
fail-closed; an uncertified bucket keeps its last certified value as its
|
||||
@@ -59,6 +68,25 @@ lifecycle discovery, and heal discovery for everyone.
|
||||
recovery-marker capabilities, but the usage-floor and observed-snapshot
|
||||
capabilities become reporting surfaces rather than proof obligations.
|
||||
|
||||
### Remaining "authority" consumers after D3 (inventory, 2026-10-19)
|
||||
|
||||
The quota path itself no longer reads scanner usage as authority for a
|
||||
durable bucket: `QuotaChecker::check_quota_with_usage_reporting` reports an
|
||||
unavailable lookup as `current_usage = None` and `map_quota_check_outcome`
|
||||
never sees `UsageUnavailable` for it. The surfaces below still carry the
|
||||
authority shape. Each is listed with its production readers, what it means
|
||||
now, the condition under which it can be retired, and the compatibility
|
||||
window. Retirement itself is separate work; none of it is required for
|
||||
correctness after this decision.
|
||||
|
||||
| Surface | Readers today | Meaning after D3 | Retirement condition | Compatibility window |
|
||||
|---|---|---|---|---|
|
||||
| `usage_snapshot_complete` (`DataUsageInfo::is_complete_bucket_usage_snapshot`) | quota-facing loader `discard_incomplete_bucket_usage`, `normalize_loaded_data_usage`, `replace_bucket_usage_memory_from_info_if_generation`, `get_bucket_usage_memory` in `crates/ecstore/src/data_usage/mod.rs`; server/pool info in `crates/ecstore/src/diagnostics/admin_server_info.rs`; `narrow_data_usage_snapshot_to_measured_buckets` in `rustfs/src/app/admin_usecase.rs`; decommission capacity in `crates/ecstore/src/core/pools.rs`; obs metrics, heal progress baseline; scanner `data_usage_info_is_cold`, `complete_scanner_cache_snapshot_plan_digest`, `reject_incomplete_snapshot` | A completeness mark for reporting (admin, Console, metrics, heal progress). For quota it only decides whether a memory value exists, which can no longer fail a durable write. | No bucket with `reservation_protocol = None` remains (lazy trigger plus admin sweep), so the loader's discard is no longer an admission gate; the mark itself stays as the reporting contract. | Until every quota is migrated, plus one release in which the snapshot-protocol admission path is still compiled. |
|
||||
| `lookup_degraded_bucket_usage_baseline` / `degraded_baseline` (`crates/ecstore/src/data_usage/mod.rs`) | `QuotaChecker::get_real_time_usage` (`crates/ecstore/src/bucket/quota/checker.rs`); `session_capacity_view` in `rustfs/src/protocols/client.rs`; fed by `load_data_usage_from_backend_with_baseline`, `backfill_degraded_baseline_from_observed`, `cache_data_usage_snapshot_result` | Admission input only for snapshot-protocol buckets; for durable buckets it only shapes the reported `current_usage`. | Same as above; afterwards it is a reporting fallback that can be dropped together with the capacity view's choice of source. | Same as above. |
|
||||
| Usage floor identity (the `scanner_epoch`/`scanner_cycle` identity of `buckets/.usage.v2.json` and `buckets/.usage.json` under `.rustfs.sys`, each with a `.bkp`; there is no separate floor object) | `persisted_usage_floor_for_startup`, `persisted_usage_floor`, `apply_persisted_usage_floor`, `reset_scanner_cycle_recovery_for_intent`, the rc.1–rc.3 `legacy_*_usage_fence` readers in `crates/scanner/src/scanner/cycle_state.rs`; `data_usage_info_has_persisted_baseline_identity` in `crates/scanner/src/scanner.rs`; `usage_snapshot_for_epoch_fence` in `crates/scanner/src/scanner/leadership.rs`; `read_data_usage_persist_baseline` in `crates/scanner/src/scanner/usage_store.rs`; `UsageSnapshotSource` resolution in `crates/ecstore/src/data_usage/mod.rs` | The scanner's own monotonic cycle fence for leadership and cycle recovery. It is no longer a quota proof: an unreadable floor stopping the leader iteration (`usage_floor_load_failed`) is a scanner-availability choice, not a quota requirement. | The quota fallback no longer reads the legacy `.usage.json` baseline; then "floor unreadable stops the scanner" can become "report and rebuild", and the rc.1–rc.3 fences (`RUSTFS_COMPAT_TODO` backlog-2102 / backlog-2122) retire on their own schedule. Cycle monotonicity stays. | One release after the compat markers' own deadline. |
|
||||
| Observed-versus-authoritative routing (`.usage.observed.json` versus `.usage.v2.json`) | decision in `classify_nsscanner_cycle`, `apply_scanner_cycle_publication_marks`, `publish_observational_snapshot`, `scanner_cycle_activity_status` (`crates/scanner/src/scanner_io.rs`); write routing in `crates/scanner/src/scanner/usage_store.rs`; readers `load_observed_data_usage_snapshot`, `select_admin_data_usage_snapshot`, `merge_partial_observation_for_admin` and the degraded-baseline backfill (`crates/ecstore/src/data_usage/mod.rs`); scanner cold-start and cache reuse; `ScannerCyclePublicationPlan::for_result` | A completeness-marked report. "Never a hard-quota authority" is moot for durable buckets; the distinction remains a reporting mark (`usage_snapshot_observational`, `converged`). | The quota fallback stops distinguishing the two objects; then the pair can collapse into one snapshot carrying the marks, after a reader-first migration of the observed-object consumers (admin, Console, obs). | Readers accept both layouts for one release before the writer changes. |
|
||||
| Publication-lease strict mode (`purpose == Authoritative && namespace_commits_pending()`) | `acquire_scanner_publication_lease_with_purpose`, `validate_scanner_publication_lease_with_purpose`, `acquire_scanner_publication_lease_guard_with_target`, `scanner_data_usage_publication_blocked` in `crates/ecstore/src/store/mod.rs`; `namespace_commits_pending` in `crates/ecstore/src/runtime/instance.rs`; `scanner_local_publication_defer_reason` in `crates/scanner/src/scanner.rs`; lease RPCs and the lease token on `RenameData`/`Delete` in `crates/protos/src/node.proto` | A pending ordinary PUT commit is a write-barrier event (the cycle is superseded or the bucket pending), not a reason to deny an authoritative lease. Data-movement and generation checks stay. | Drop the pending-commit rejection and its two revalidation twins, and the pending-commit term of the local barrier; keep the movement fence and the lease token on writes. The local barrier's term is handled by the scanner cycle-termination work; the remote lease follows once that lands. | Mixed versions are safe: an old peer still rejects and the coordinator treats that as retryable (`scanner_publication_lease_error_is_retryable`), so the removal rolls out node by node. |
|
||||
|
||||
## Superseded decision (2026-09-03)
|
||||
|
||||
RustFS keeps scanner data usage as authoritative for quota admission.
|
||||
|
||||
@@ -116,3 +116,69 @@ bound, not a best-effort token bucket.
|
||||
Unused credit is intentionally retained until the quota epoch changes or a
|
||||
future reconciliation operation returns it, trading some capacity utilization
|
||||
for a simple crash-safe upper bound.
|
||||
|
||||
## Quotas saved before the reservation protocols
|
||||
|
||||
A quota configuration without `reservation_protocol` (saved by a release
|
||||
before protocol v1) is still admitted from scanner usage snapshots: its writes
|
||||
need an authoritative or degraded usage value and fail closed with a
|
||||
retryable `503` when neither exists. Decision D3
|
||||
([scanner-usage-authority-decision.md](../architecture/scanner-usage-authority-decision.md))
|
||||
moves such buckets onto the sharded durable protocol by re-saving the same
|
||||
limit and type with `reservation_protocol: 2`; `created_at` is preserved.
|
||||
|
||||
Two triggers exist and both go through the same guarded save:
|
||||
|
||||
- Lazily, on the first quota read of such a bucket through the instance's
|
||||
registered bucket metadata system (every PUT, COPY, multipart completion
|
||||
and admin quota read). The read answers with the configuration it found and
|
||||
schedules the re-save in the background; it is never blocked or failed by
|
||||
the attempt. An attempt that does not settle the bucket is retried no
|
||||
sooner than five minutes later.
|
||||
- In bulk, with `POST /rustfs/admin/v3/quota-migration`
|
||||
(`admin:SetBucketQuota`). The sweep visits every bucket and answers with a
|
||||
per-bucket report:
|
||||
|
||||
```json
|
||||
{
|
||||
"target_reservation_protocol": 2,
|
||||
"migrated": 1, "already_durable": 3, "no_quota": 10, "skipped": 0, "failed": 0,
|
||||
"buckets": [
|
||||
{"bucket": "photos", "outcome": "migrated", "reservation_protocol": 2, "updated_at": "2026-10-19T02:14:07Z"},
|
||||
{"bucket": "logs", "outcome": "already_durable", "reservation_protocol": 1},
|
||||
{"bucket": "tmp", "outcome": "no_quota"}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
`outcome` is one of `migrated`, `already_durable`, `no_quota`, `skipped`
|
||||
(`reason`: `unsupported_reservation_protocol`, `quota_revision_changed`,
|
||||
`bucket_incarnation_changed`) or `failed` (`reason` carries the error). A
|
||||
failed bucket does not stop the sweep; run the command again after fixing the
|
||||
cause.
|
||||
|
||||
The save keeps the admin save's preconditions: the whole-fleet capability
|
||||
proof must be live (a mixed-version fleet fails the attempt rather than
|
||||
exposing protocol v2 to a node that would fail closed on it), the bucket
|
||||
incarnation must still match, and the quota revision must be the one that was
|
||||
read. A quota edited in between is reported as `quota_revision_changed` and
|
||||
left for the next attempt. A protocol this node does not understand is never
|
||||
rewritten. Durable quotas (protocol v1 or v2) are not touched; the admin
|
||||
default for a new quota remains protocol v1.
|
||||
|
||||
Each attempt logs one `quota_reservation_protocol_migration` event
|
||||
(`trigger` is `lazy` or `admin`, `result` is `migrated`, `noop`, `skipped` or
|
||||
`failed`) and increments
|
||||
`rustfs_quota_reservation_protocol_migrations_total` with the same labels.
|
||||
|
||||
## Usage reporting for durable quotas
|
||||
|
||||
For a bucket on protocol v1 or v2 the request-path quota check and the admin
|
||||
quota endpoints read scanner usage for reporting only. When no usage is
|
||||
available, `GET /rustfs/admin/v3/quota-stats/{bucket}` and
|
||||
`POST /rustfs/admin/v3/quota-check/{bucket}` answer `200` with
|
||||
`current_usage: null` (and no `remaining_quota` or `usage_percentage`) instead
|
||||
of `503`, and object writes are admitted by the ledger alone. Each such
|
||||
lookup logs a `quota_usage_report` event at debug level and increments
|
||||
`rustfs_quota_usage_report_unavailable_total`. A bucket without
|
||||
`reservation_protocol` keeps the previous behaviour until it is migrated.
|
||||
|
||||
@@ -19,8 +19,11 @@ use crate::admin::handlers::site_replication::site_replication_bucket_meta_hook;
|
||||
use crate::admin::router::{AdminOperation, Operation, S3Router};
|
||||
use crate::admin::runtime_sources::{current_bucket_metadata_handle, current_object_store_handle};
|
||||
use crate::admin::storage_api::bucket::metadata_sys::{self, BucketMetadataSys};
|
||||
use crate::admin::storage_api::bucket::quota::checker::QuotaChecker;
|
||||
use crate::admin::storage_api::bucket::quota::{BucketQuota, QuotaError, QuotaOperation};
|
||||
use crate::admin::storage_api::bucket::quota::checker::{QuotaChecker, record_reservation_protocol_migration};
|
||||
use crate::admin::storage_api::bucket::quota::{
|
||||
BucketQuota, QUOTA_RESERVATION_PROTOCOL_V2, QuotaError, QuotaOperation, QuotaProtocolMigration,
|
||||
};
|
||||
use crate::admin::storage_api::contract::bucket::{BucketOperations as _, BucketOptions};
|
||||
use crate::auth::{check_key_valid, get_session_token};
|
||||
use crate::error::ApiError;
|
||||
use crate::server::ADMIN_PREFIX;
|
||||
@@ -33,6 +36,7 @@ use s3s::{Body, S3Error, S3Request, S3Response, S3Result, s3_error};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json;
|
||||
use std::sync::Arc;
|
||||
use time::format_description::well_known::Rfc3339;
|
||||
use tokio::sync::RwLock;
|
||||
use tracing::{debug, info, warn};
|
||||
use url::form_urlencoded;
|
||||
@@ -158,7 +162,9 @@ pub struct BucketQuotaResponse {
|
||||
pub struct BucketQuotaStats {
|
||||
pub bucket: String,
|
||||
pub quota_limit: Option<u64>,
|
||||
pub current_usage: u64,
|
||||
/// `null` when the bucket is admitted by the reservation ledger and no
|
||||
/// scanner usage is available; never reported as zero.
|
||||
pub current_usage: Option<u64>,
|
||||
pub remaining_quota: Option<u64>,
|
||||
pub usage_percentage: Option<f64>,
|
||||
}
|
||||
@@ -175,17 +181,107 @@ pub struct CheckQuotaResponse {
|
||||
pub operation_type: String,
|
||||
pub operation_size: u64,
|
||||
pub allowed: bool,
|
||||
pub current_usage: u64,
|
||||
/// `null` when the bucket is admitted by the reservation ledger and no
|
||||
/// scanner usage is available; never reported as zero.
|
||||
pub current_usage: Option<u64>,
|
||||
pub quota_limit: Option<u64>,
|
||||
pub remaining_quota: Option<u64>,
|
||||
}
|
||||
|
||||
/// One bucket of the reservation protocol migration report.
|
||||
#[derive(Debug, Serialize)]
|
||||
pub struct BucketQuotaProtocolMigrationEntry {
|
||||
pub bucket: String,
|
||||
/// `migrated`, `already_durable`, `no_quota`, `skipped` or `failed`.
|
||||
pub outcome: &'static str,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub reservation_protocol: Option<u32>,
|
||||
/// The skip reason or the failure message.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub reason: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub updated_at: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Serialize)]
|
||||
pub struct BucketQuotaProtocolMigrationReport {
|
||||
pub target_reservation_protocol: u32,
|
||||
pub migrated: usize,
|
||||
pub already_durable: usize,
|
||||
pub no_quota: usize,
|
||||
pub skipped: usize,
|
||||
pub failed: usize,
|
||||
pub buckets: Vec<BucketQuotaProtocolMigrationEntry>,
|
||||
}
|
||||
|
||||
impl BucketQuotaProtocolMigrationReport {
|
||||
fn record(&mut self, bucket: String, outcome: &Result<QuotaProtocolMigration, QuotaError>) {
|
||||
let entry = match outcome {
|
||||
Ok(QuotaProtocolMigration::Migrated {
|
||||
reservation_protocol,
|
||||
updated_at,
|
||||
}) => {
|
||||
self.migrated += 1;
|
||||
BucketQuotaProtocolMigrationEntry {
|
||||
bucket,
|
||||
outcome: "migrated",
|
||||
reservation_protocol: Some(*reservation_protocol),
|
||||
reason: None,
|
||||
updated_at: updated_at.format(&Rfc3339).ok(),
|
||||
}
|
||||
}
|
||||
Ok(QuotaProtocolMigration::AlreadyDurable { reservation_protocol }) => {
|
||||
self.already_durable += 1;
|
||||
BucketQuotaProtocolMigrationEntry {
|
||||
bucket,
|
||||
outcome: "already_durable",
|
||||
reservation_protocol: Some(*reservation_protocol),
|
||||
reason: None,
|
||||
updated_at: None,
|
||||
}
|
||||
}
|
||||
Ok(QuotaProtocolMigration::NoQuota) => {
|
||||
self.no_quota += 1;
|
||||
BucketQuotaProtocolMigrationEntry {
|
||||
bucket,
|
||||
outcome: "no_quota",
|
||||
reservation_protocol: None,
|
||||
reason: None,
|
||||
updated_at: None,
|
||||
}
|
||||
}
|
||||
Ok(QuotaProtocolMigration::Skipped { reason }) => {
|
||||
self.skipped += 1;
|
||||
BucketQuotaProtocolMigrationEntry {
|
||||
bucket,
|
||||
outcome: "skipped",
|
||||
reservation_protocol: None,
|
||||
reason: Some(reason.as_str().to_string()),
|
||||
updated_at: None,
|
||||
}
|
||||
}
|
||||
Err(err) => {
|
||||
self.failed += 1;
|
||||
BucketQuotaProtocolMigrationEntry {
|
||||
bucket,
|
||||
outcome: "failed",
|
||||
reservation_protocol: None,
|
||||
reason: Some(err.to_string()),
|
||||
updated_at: None,
|
||||
}
|
||||
}
|
||||
};
|
||||
self.buckets.push(entry);
|
||||
}
|
||||
}
|
||||
|
||||
/// Quota management handlers
|
||||
pub struct SetBucketQuotaHandler;
|
||||
pub struct GetBucketQuotaHandler;
|
||||
pub struct ClearBucketQuotaHandler;
|
||||
pub struct GetBucketQuotaStatsHandler;
|
||||
pub struct CheckBucketQuotaHandler;
|
||||
pub struct MigrateBucketQuotaProtocolHandler;
|
||||
|
||||
fn bucket_metadata_from_context() -> Option<Arc<RwLock<BucketMetadataSys>>> {
|
||||
current_bucket_metadata_handle()
|
||||
@@ -249,6 +345,12 @@ pub fn register_quota_route(r: &mut S3Router<AdminOperation>) -> std::io::Result
|
||||
AdminOperation(&CheckBucketQuotaHandler {}),
|
||||
)?;
|
||||
|
||||
r.insert(
|
||||
Method::POST,
|
||||
format!("{}{}", ADMIN_PREFIX, "/v3/quota-migration").as_str(),
|
||||
AdminOperation(&MigrateBucketQuotaProtocolHandler {}),
|
||||
)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -614,21 +716,17 @@ impl Operation for GetBucketQuotaStatsHandler {
|
||||
|
||||
let quota_checker = QuotaChecker::new(metadata_sys_lock.clone());
|
||||
|
||||
let (quota, current_usage_opt) = quota_checker
|
||||
let (quota, current_usage) = quota_checker
|
||||
.get_quota_stats(&bucket)
|
||||
.await
|
||||
.map_err(|error| map_quota_query_error(error, &bucket, "get quota stats"))?;
|
||||
|
||||
let current_usage = current_usage_opt.unwrap_or(0);
|
||||
let usage_percentage = quota.quota.and_then(|limit| {
|
||||
if limit == 0 {
|
||||
None
|
||||
} else {
|
||||
Some((current_usage as f64 / limit as f64) * 100.0)
|
||||
}
|
||||
});
|
||||
let usage_percentage = match (quota.quota, current_usage) {
|
||||
(Some(limit), Some(current_usage)) if limit > 0 => Some((current_usage as f64 / limit as f64) * 100.0),
|
||||
_ => None,
|
||||
};
|
||||
|
||||
let remaining_quota = quota.get_remaining_quota(current_usage);
|
||||
let remaining_quota = current_usage.and_then(|current_usage| quota.get_remaining_quota(current_usage));
|
||||
|
||||
let response = BucketQuotaStats {
|
||||
bucket,
|
||||
@@ -729,7 +827,7 @@ impl Operation for CheckBucketQuotaHandler {
|
||||
operation_type: request.operation_type,
|
||||
operation_size: request.operation_size,
|
||||
allowed: result.allowed,
|
||||
current_usage: result.current_usage.unwrap_or(0),
|
||||
current_usage: result.current_usage,
|
||||
quota_limit: result.quota_limit,
|
||||
remaining_quota: result.remaining,
|
||||
};
|
||||
@@ -741,6 +839,77 @@ impl Operation for CheckBucketQuotaHandler {
|
||||
}
|
||||
}
|
||||
|
||||
/// `POST /rustfs/admin/v3/quota-migration`: re-save every quota that predates
|
||||
/// the reservation protocols under the sharded durable protocol and report
|
||||
/// each bucket (rustfs/backlog#2904). The limit and type are unchanged; a
|
||||
/// bucket that fails is reported, not retried, and never stops the sweep.
|
||||
#[async_trait::async_trait]
|
||||
impl Operation for MigrateBucketQuotaProtocolHandler {
|
||||
#[tracing::instrument(skip_all)]
|
||||
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
warn!(
|
||||
event = EVENT_ADMIN_QUOTA_STATE,
|
||||
component = LOG_COMPONENT_ADMIN,
|
||||
subsystem = LOG_SUBSYSTEM_QUOTA,
|
||||
action = "migrate_bucket_quota_protocol",
|
||||
state = "requested",
|
||||
"admin quota state"
|
||||
);
|
||||
|
||||
let Some(ref cred) = req.credentials else {
|
||||
return Err(s3_error!(InvalidRequest, "authentication required"));
|
||||
};
|
||||
|
||||
let (cred, owner) =
|
||||
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
|
||||
|
||||
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
|
||||
validate_admin_request(
|
||||
&req.headers,
|
||||
&cred,
|
||||
owner,
|
||||
false,
|
||||
vec![Action::AdminAction(AdminAction::SetBucketQuotaAdminAction)],
|
||||
remote_addr,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let store = current_object_store_handle().ok_or_else(|| s3_error!(InternalError, "object store is not initialized"))?;
|
||||
let metadata_sys_lock = bucket_metadata_from_context()
|
||||
.ok_or_else(|| s3_error!(InternalError, "{}", rustfs_config::QUOTA_METADATA_SYSTEM_ERROR_MSG))?;
|
||||
let buckets = store.list_bucket(&BucketOptions::default()).await.map_err(ApiError::from)?;
|
||||
|
||||
let quota_checker = QuotaChecker::new(metadata_sys_lock);
|
||||
let mut report = BucketQuotaProtocolMigrationReport {
|
||||
target_reservation_protocol: QUOTA_RESERVATION_PROTOCOL_V2,
|
||||
..Default::default()
|
||||
};
|
||||
for bucket in buckets {
|
||||
let outcome = quota_checker.migrate_reservation_protocol(&bucket.name).await;
|
||||
record_reservation_protocol_migration(&bucket.name, "admin", &outcome);
|
||||
report.record(bucket.name, &outcome);
|
||||
}
|
||||
|
||||
info!(
|
||||
event = EVENT_ADMIN_QUOTA_STATE,
|
||||
component = LOG_COMPONENT_ADMIN,
|
||||
subsystem = LOG_SUBSYSTEM_QUOTA,
|
||||
action = "migrate_bucket_quota_protocol",
|
||||
state = "completed",
|
||||
migrated = report.migrated,
|
||||
already_durable = report.already_durable,
|
||||
no_quota = report.no_quota,
|
||||
skipped = report.skipped,
|
||||
failed = report.failed,
|
||||
"admin quota state"
|
||||
);
|
||||
|
||||
let json = serde_json::to_string(&report).map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?;
|
||||
|
||||
Ok(S3Response::new((StatusCode::OK, Body::from(json))))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -801,6 +970,80 @@ mod tests {
|
||||
assert_eq!(error.code(), &s3s::S3ErrorCode::ServiceUnavailable);
|
||||
}
|
||||
|
||||
/// rustfs/backlog#2904: unknown usage on a ledger-admitted bucket is
|
||||
/// reported as `null`, never as zero.
|
||||
#[test]
|
||||
fn quota_stats_report_unknown_usage_as_null() {
|
||||
let stats = BucketQuotaStats {
|
||||
bucket: "test-bucket".to_string(),
|
||||
quota_limit: Some(2048),
|
||||
current_usage: None,
|
||||
remaining_quota: None,
|
||||
usage_percentage: None,
|
||||
};
|
||||
let value = serde_json::to_value(&stats).expect("stats should serialize");
|
||||
assert!(value["current_usage"].is_null());
|
||||
assert!(value["remaining_quota"].is_null());
|
||||
|
||||
let check = CheckQuotaResponse {
|
||||
bucket: "test-bucket".to_string(),
|
||||
operation_type: "PUT".to_string(),
|
||||
operation_size: 512,
|
||||
allowed: true,
|
||||
current_usage: None,
|
||||
quota_limit: Some(2048),
|
||||
remaining_quota: None,
|
||||
};
|
||||
let value = serde_json::to_value(&check).expect("check response should serialize");
|
||||
assert!(value["current_usage"].is_null());
|
||||
assert_eq!(value["allowed"], serde_json::Value::Bool(true));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn quota_protocol_migration_report_counts_every_outcome() {
|
||||
use crate::admin::storage_api::bucket::quota::QuotaMigrationSkipReason;
|
||||
|
||||
let mut report = BucketQuotaProtocolMigrationReport {
|
||||
target_reservation_protocol: QUOTA_RESERVATION_PROTOCOL_V2,
|
||||
..Default::default()
|
||||
};
|
||||
report.record(
|
||||
"a".to_string(),
|
||||
&Ok(QuotaProtocolMigration::Migrated {
|
||||
reservation_protocol: QUOTA_RESERVATION_PROTOCOL_V2,
|
||||
updated_at: time::OffsetDateTime::UNIX_EPOCH,
|
||||
}),
|
||||
);
|
||||
report.record("b".to_string(), &Ok(QuotaProtocolMigration::AlreadyDurable { reservation_protocol: 1 }));
|
||||
report.record("c".to_string(), &Ok(QuotaProtocolMigration::NoQuota));
|
||||
report.record(
|
||||
"d".to_string(),
|
||||
&Ok(QuotaProtocolMigration::Skipped {
|
||||
reason: QuotaMigrationSkipReason::QuotaRevisionChanged,
|
||||
}),
|
||||
);
|
||||
report.record("e".to_string(), &Err(QuotaError::UsageUnavailable { bucket: "e".to_string() }));
|
||||
|
||||
assert_eq!(
|
||||
(report.migrated, report.already_durable, report.no_quota, report.skipped, report.failed),
|
||||
(1, 1, 1, 1, 1)
|
||||
);
|
||||
let value = serde_json::to_value(&report).expect("report should serialize");
|
||||
assert_eq!(value["target_reservation_protocol"], 2);
|
||||
assert_eq!(value["buckets"][0]["outcome"], "migrated");
|
||||
assert_eq!(value["buckets"][0]["reservation_protocol"], 2);
|
||||
assert_eq!(value["buckets"][0]["updated_at"], "1970-01-01T00:00:00Z");
|
||||
assert_eq!(value["buckets"][1]["reservation_protocol"], 1);
|
||||
assert!(value["buckets"][2].get("reason").is_none());
|
||||
assert_eq!(value["buckets"][3]["reason"], "quota_revision_changed");
|
||||
assert_eq!(value["buckets"][4]["outcome"], "failed");
|
||||
assert!(
|
||||
value["buckets"][4]["reason"]
|
||||
.as_str()
|
||||
.is_some_and(|reason| reason.contains("e"))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_quota_operation_parsing() {
|
||||
let parse_operation = |operation: &str| match operation.to_uppercase().as_str() {
|
||||
|
||||
@@ -395,6 +395,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
|
||||
SET_BUCKET_QUOTA,
|
||||
RouteRiskLevel::High,
|
||||
),
|
||||
admin(
|
||||
HttpMethod::Post,
|
||||
"/rustfs/admin/v3/quota-migration",
|
||||
SET_BUCKET_QUOTA,
|
||||
RouteRiskLevel::High,
|
||||
),
|
||||
admin(
|
||||
HttpMethod::Put,
|
||||
"/rustfs/admin/v3/bucket-durability/{bucket}",
|
||||
|
||||
@@ -263,6 +263,7 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
|
||||
admin_route_sample(Method::DELETE, "/v3/quota/{bucket}", "/v3/quota/test-bucket"),
|
||||
admin_route_sample(Method::GET, "/v3/quota-stats/{bucket}", "/v3/quota-stats/test-bucket"),
|
||||
admin_route_sample(Method::POST, "/v3/quota-check/{bucket}", "/v3/quota-check/test-bucket"),
|
||||
admin_route(Method::POST, "/v3/quota-migration"),
|
||||
admin_route_sample(Method::PUT, "/v3/bucket-durability/{bucket}", "/v3/bucket-durability/test-bucket"),
|
||||
admin_route_sample(Method::GET, "/v3/bucket-durability/{bucket}", "/v3/bucket-durability/test-bucket"),
|
||||
admin_route_sample(Method::DELETE, "/v3/bucket-durability/{bucket}", "/v3/bucket-durability/test-bucket"),
|
||||
@@ -1382,6 +1383,7 @@ fn test_register_routes_cover_representative_admin_paths() {
|
||||
assert_route(&router, Method::GET, &admin_path("/v3/get-bucket-quota"));
|
||||
assert_route(&router, Method::PUT, &admin_path("/v3/quota/test-bucket"));
|
||||
assert_route(&router, Method::GET, &admin_path("/v3/quota-stats/test-bucket"));
|
||||
assert_route(&router, Method::POST, &admin_path("/v3/quota-migration"));
|
||||
assert_route(&router, Method::PUT, &admin_path("/v3/bucket-durability/test-bucket"));
|
||||
assert_route(&router, Method::GET, &admin_path("/v3/bucket-durability/test-bucket"));
|
||||
assert_route(&router, Method::DELETE, &admin_path("/v3/bucket-durability/test-bucket"));
|
||||
|
||||
@@ -532,11 +532,15 @@ pub(crate) mod metadata_sys {
|
||||
pub(crate) mod quota {
|
||||
pub(crate) mod checker {
|
||||
pub(crate) type QuotaChecker = super::super::ecstore_bucket::quota::checker::QuotaChecker;
|
||||
pub(crate) use super::super::ecstore_bucket::quota::checker::record_reservation_protocol_migration;
|
||||
}
|
||||
|
||||
pub(crate) type BucketQuota = super::ecstore_bucket::quota::BucketQuota;
|
||||
pub(crate) type QuotaError = super::ecstore_bucket::quota::QuotaError;
|
||||
pub(crate) type QuotaOperation = super::ecstore_bucket::quota::QuotaOperation;
|
||||
pub(crate) type QuotaProtocolMigration = super::ecstore_bucket::quota::QuotaProtocolMigration;
|
||||
#[cfg(test)]
|
||||
pub(crate) type QuotaMigrationSkipReason = super::ecstore_bucket::quota::QuotaMigrationSkipReason;
|
||||
pub(crate) use super::ecstore_bucket::quota::{QUOTA_RESERVATION_PROTOCOL_V1, QUOTA_RESERVATION_PROTOCOL_V2};
|
||||
}
|
||||
|
||||
|
||||
@@ -4113,6 +4113,62 @@ mod tests {
|
||||
assert!(!body_polled.load(Ordering::Acquire), "legacy quota rejection must not consume the body");
|
||||
}
|
||||
|
||||
/// Regression (rustfs/backlog#2904, decision D3 in rustfs/backlog#2872): a
|
||||
/// bucket admitted by the durable reservation ledger must accept a write
|
||||
/// even when no scanner usage baseline exists anywhere, and the ledger
|
||||
/// alone still refuses an over-limit write. Before the fix the checker's
|
||||
/// `UsageUnavailable` was mapped to a retryable 503 for every PUT.
|
||||
#[tokio::test]
|
||||
#[serial_test::serial]
|
||||
async fn durable_quota_put_succeeds_without_any_usage_baseline() {
|
||||
use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
|
||||
|
||||
const LIMIT: u64 = 1 << 20;
|
||||
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!("durable-no-usage-{}", Uuid::new_v4().simple());
|
||||
store
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("create durable quota test bucket");
|
||||
// Deliberately no usage seed: neither the memory cache nor a persisted
|
||||
// snapshot knows this bucket, so the scanner-side lookup fails.
|
||||
let metadata_sys = DefaultObjectUsecase::from_global()
|
||||
.bucket_metadata_sys()
|
||||
.expect("test app context should expose bucket metadata");
|
||||
QuotaChecker::new(metadata_sys)
|
||||
.set_quota_config(&bucket, BucketQuota::new(Some(LIMIT)))
|
||||
.await
|
||||
.expect("configure durable quota");
|
||||
|
||||
let payload = Bytes::from(vec![0x42; 1024]);
|
||||
let input = PutObjectInput::builder()
|
||||
.bucket(bucket.clone())
|
||||
.key("fits".to_string())
|
||||
.body(Some(StreamingBlob::from(s3s::Body::from(payload.clone()))))
|
||||
.content_length(Some(i64::try_from(payload.len()).expect("test payload length must fit i64")))
|
||||
.build()
|
||||
.expect("PUT input should build");
|
||||
DefaultObjectUsecase::from_global()
|
||||
.execute_put_object(&FS::new(), build_request(input, Method::PUT))
|
||||
.await
|
||||
.expect("a ledger-admitted bucket must accept a write without scanner usage");
|
||||
|
||||
let too_large = Bytes::from(vec![0x42; (LIMIT + 1) as usize]);
|
||||
let input = PutObjectInput::builder()
|
||||
.bucket(bucket)
|
||||
.key("too-large".to_string())
|
||||
.body(Some(StreamingBlob::from(s3s::Body::from(too_large.clone()))))
|
||||
.content_length(Some(i64::try_from(too_large.len()).expect("test payload length must fit i64")))
|
||||
.build()
|
||||
.expect("PUT input should build");
|
||||
let err = DefaultObjectUsecase::from_global()
|
||||
.execute_put_object(&FS::new(), build_request(input, Method::PUT))
|
||||
.await
|
||||
.expect_err("the ledger must still reject a write above the quota limit");
|
||||
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial]
|
||||
async fn concurrent_puts_share_durable_bucket_quota_reservations() {
|
||||
|
||||
@@ -198,7 +198,7 @@ pub(super) fn request_body_is_aws_chunked_framed(headers: &HeaderMap) -> bool {
|
||||
|
||||
/// Map a bucket-quota checker outcome onto the S3 admission result.
|
||||
///
|
||||
/// Hard is the only supported quota type, so a checker fault (bucket-config read, config parse, or usage lookup) must fail closed rather than admit the write: allowing it would silently bypass a configured hard quota. The no-quota happy path never reaches the error arm — `QuotaChecker::check_quota` returns `Ok(allowed)` via the zero-extra-I/O fast path when no quota is configured, so failing closed here cannot penalise buckets without a quota. A fault surfaces as a retryable `ServiceUnavailable` and is counted; the client-facing message stays generic so internal config/usage details are not leaked.
|
||||
/// Hard is the only supported quota type, so a checker fault (bucket-config read, config parse, or usage lookup) must fail closed rather than admit the write: allowing it would silently bypass a configured hard quota. The no-quota happy path never reaches the error arm — `QuotaChecker::check_quota` returns `Ok(allowed)` via the zero-extra-I/O fast path when no quota is configured, so failing closed here cannot penalise buckets without a quota. A bucket admitted by the durable reservation ledger never reaches it for an unavailable usage lookup either: the checker reports that usage as absent (`current_usage = None`) and the ledger settles admission at commit (rustfs/backlog#2904); only the snapshot protocol still fails closed on it. A fault surfaces as a retryable `ServiceUnavailable` and is counted; the client-facing message stays generic so internal config/usage details are not leaked.
|
||||
pub(crate) fn map_quota_check_outcome(bucket: &str, outcome: Result<QuotaCheckResult, QuotaError>) -> S3Result<QuotaCheckResult> {
|
||||
match outcome {
|
||||
Ok(result) if !result.allowed => Err(S3Error::with_message(
|
||||
@@ -2184,15 +2184,39 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn quota_admission_fails_closed_on_unknown_authoritative_usage() {
|
||||
// Only the snapshot protocol still surfaces this: the checker never
|
||||
// returns it for a bucket the durable reservation ledger admits.
|
||||
let err = map_quota_check_outcome(
|
||||
"bucket",
|
||||
Err(QuotaError::UsageUnavailable {
|
||||
bucket: "bucket".to_string(),
|
||||
}),
|
||||
)
|
||||
.expect_err("unknown authoritative usage must not admit a quota-controlled write");
|
||||
.expect_err("unknown authoritative usage must not admit a snapshot-protocol write");
|
||||
assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable);
|
||||
}
|
||||
|
||||
/// rustfs/backlog#2904: a durable-quota check whose scanner usage is
|
||||
/// unknown admits the write at the handler and leaves the decision to the
|
||||
/// ledger; nothing on the handler path treats the absent usage as zero.
|
||||
#[test]
|
||||
fn durable_quota_without_scanner_usage_admits_the_write() {
|
||||
let unknown_usage = QuotaCheckResult {
|
||||
allowed: true,
|
||||
current_usage: None,
|
||||
quota_limit: Some(2048),
|
||||
operation_size: 512,
|
||||
remaining: None,
|
||||
uses_durable_reservations: true,
|
||||
};
|
||||
let result = map_quota_check_outcome("bucket", Ok(unknown_usage)).expect("the ledger settles admission at commit");
|
||||
assert_eq!(result.current_usage, None);
|
||||
|
||||
let mut opts = ObjectOptions::default();
|
||||
apply_quota_admission(&mut opts, &result).expect("durable admission carries no snapshot into the write");
|
||||
assert!(opts.quota_admission.is_none());
|
||||
ensure_object_size_within_quota(&result, 4096).expect("unknown usage cannot pre-reject an object");
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
Reference in New Issue
Block a user