From 9a1f3d2c4cd5dfe3335a7af142e317f4d0dffcd8 Mon Sep 17 00:00:00 2001 From: Chris Date: Mon, 19 Oct 2026 22:45:08 +0900 Subject: [PATCH] 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 --- crates/ecstore/src/api/mod.rs | 4 +- crates/ecstore/src/bucket/metadata_sys.rs | 46 ++ crates/ecstore/src/bucket/quota/checker.rs | 656 +++++++++++++++++- crates/ecstore/src/bucket/quota/mod.rs | 59 ++ .../scanner-usage-authority-decision.md | 30 +- .../durable-quota-reservation-v2.md | 66 ++ rustfs/src/admin/handlers/quota.rs | 273 +++++++- rustfs/src/admin/route_policy.rs | 6 + rustfs/src/admin/route_registration_test.rs | 2 + rustfs/src/admin/storage_api.rs | 4 + rustfs/src/app/object/put.rs | 56 ++ rustfs/src/app/object/shared.rs | 28 +- 12 files changed, 1192 insertions(+), 38 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 2d7a2efac..ef3793625 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -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}; } } diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index 4a56dceb3..cb2ad0fb3 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -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>, + bucket: &str, + data: Vec, + expected_incarnation_id: Uuid, + expected_quota_revision: OffsetDateTime, + proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken, +) -> Result { + 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, diff --git a/crates/ecstore/src/bucket/quota/checker.rs b/crates/ecstore/src/bucket/quota/checker.rs index 8e313c09a..d6b3192ad 100644 --- a/crates/ecstore/src/bucket/quota/checker.rs +++ b/crates/ecstore/src/bucket/quota/checker.rs @@ -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>, @@ -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 { + 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 { + 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> { + static ATTEMPTS: OnceLock>> = 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, +) { + 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, Arc, Arc>) + { + 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, bucket: &str, quota_json: Vec) { + 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 { + format!(r#"{{"quota":{limit},"quota_type":"Hard","created_at":"2024-01-01T00:00:00Z"}}"#).into_bytes() + } + + async fn persisted_quota(store: &Arc, 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 { diff --git a/crates/ecstore/src/bucket/quota/mod.rs b/crates/ecstore/src/bucket/quota/mod.rs index b57494250..1187327e1 100644 --- a/crates/ecstore/src/bucket/quota/mod.rs +++ b/crates/ecstore/src/bucket/quota/mod.rs @@ -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.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}")] diff --git a/docs/architecture/scanner-usage-authority-decision.md b/docs/architecture/scanner-usage-authority-decision.md index ed0a8038b..edd7ed8d9 100644 --- a/docs/architecture/scanner-usage-authority-decision.md +++ b/docs/architecture/scanner-usage-authority-decision.md @@ -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. diff --git a/docs/operations/durable-quota-reservation-v2.md b/docs/operations/durable-quota-reservation-v2.md index 250cc84ff..862ed9f70 100644 --- a/docs/operations/durable-quota-reservation-v2.md +++ b/docs/operations/durable-quota-reservation-v2.md @@ -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. diff --git a/rustfs/src/admin/handlers/quota.rs b/rustfs/src/admin/handlers/quota.rs index 8c6f2e18b..892517d8e 100644 --- a/rustfs/src/admin/handlers/quota.rs +++ b/rustfs/src/admin/handlers/quota.rs @@ -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, - 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, pub remaining_quota: Option, pub usage_percentage: Option, } @@ -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, pub quota_limit: Option, pub remaining_quota: Option, } +/// 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, + /// The skip reason or the failure message. + #[serde(skip_serializing_if = "Option::is_none")] + pub reason: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub updated_at: Option, +} + +#[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, +} + +impl BucketQuotaProtocolMigrationReport { + fn record(&mut self, bucket: String, outcome: &Result) { + 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>> { current_bucket_metadata_handle() @@ -249,6 +345,12 @@ pub fn register_quota_route(r: &mut S3Router) -> 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, _params: Params<'_, '_>) -> S3Result> { + 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::>().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() { diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 8acd5f313..be233427d 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -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}", diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index ac24a3d6f..130098d91 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -263,6 +263,7 @@ fn expected_admin_route_matrix() -> Vec { 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")); diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 08ea72948..352357a4e 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -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}; } diff --git a/rustfs/src/app/object/put.rs b/rustfs/src/app/object/put.rs index 757238025..e88e4f49a 100644 --- a/rustfs/src/app/object/put.rs +++ b/rustfs/src/app/object/put.rs @@ -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() { diff --git a/rustfs/src/app/object/shared.rs b/rustfs/src/app/object/shared.rs index ac8ddeddf..e4d7cc295 100644 --- a/rustfs/src/app/object/shared.rs +++ b/rustfs/src/app/object/shared.rs @@ -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) -> S3Result { 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)]