diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index e45d56f09..4d60b652b 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -240,7 +240,10 @@ pub mod bucket { } pub mod quota { - pub use crate::bucket::quota::{BucketQuota, QuotaCheckResult, QuotaError, QuotaOperation}; + pub use crate::bucket::quota::{ + BucketQuota, QUOTA_RESERVATION_PROTOCOL_V1, QUOTA_RESERVATION_PROTOCOL_V2, QuotaCheckResult, QuotaError, + QuotaOperation, + }; pub mod checker { pub use crate::bucket::quota::checker::QuotaChecker; diff --git a/crates/ecstore/src/bucket/quota/mod.rs b/crates/ecstore/src/bucket/quota/mod.rs index 3fed86728..b57494250 100644 --- a/crates/ecstore/src/bucket/quota/mod.rs +++ b/crates/ecstore/src/bucket/quota/mod.rs @@ -32,7 +32,10 @@ pub enum QuotaType { Hard, } -pub(crate) const QUOTA_RESERVATION_PROTOCOL_V1: u32 = 1; +pub const QUOTA_RESERVATION_PROTOCOL_V1: u32 = 1; +/// Sharded durable reservations. Older nodes reject this protocol instead of +/// silently falling back to the single-ledger protocol. +pub const QUOTA_RESERVATION_PROTOCOL_V2: u32 = 2; /// Bucket quota configuration. quota_type defaults to Hard when omitted. #[derive(Debug, Default, Clone, PartialEq)] @@ -89,7 +92,10 @@ impl<'de> Deserialize<'de> for BucketQuota { D: Deserializer<'de>, { let wire = BucketQuotaWire::deserialize(deserializer)?; - let quota = if wire.reservation_protocol == Some(QUOTA_RESERVATION_PROTOCOL_V1) { + let quota = if matches!( + wire.reservation_protocol, + Some(QUOTA_RESERVATION_PROTOCOL_V1) | Some(QUOTA_RESERVATION_PROTOCOL_V2) + ) { Some( wire.reservation_quota .ok_or_else(|| D::Error::custom("reservation_quota is required for reservation protocol v1"))?, @@ -129,17 +135,33 @@ impl BucketQuota { } } + /// Creates a quota using the sharded durable reservation protocol. + /// Deployment must complete the mixed-version rollout before enabling it; + /// older nodes fail closed when they see protocol v2. + pub fn new_sharded(quota: u64) -> Self { + let mut config = Self::new(Some(quota)); + config.reservation_protocol = Some(QUOTA_RESERVATION_PROTOCOL_V2); + config + } + pub fn get_quota_limit(&self) -> Option { self.quota } pub fn uses_durable_reservations(&self) -> bool { - self.reservation_protocol == Some(QUOTA_RESERVATION_PROTOCOL_V1) + matches!( + self.reservation_protocol, + Some(QUOTA_RESERVATION_PROTOCOL_V1) | Some(QUOTA_RESERVATION_PROTOCOL_V2) + ) + } + + pub fn uses_sharded_reservations(&self) -> bool { + self.reservation_protocol == Some(QUOTA_RESERVATION_PROTOCOL_V2) } pub fn has_unsupported_reservation_protocol(&self) -> bool { self.reservation_protocol - .is_some_and(|version| version != QUOTA_RESERVATION_PROTOCOL_V1) + .is_some_and(|version| !matches!(version, QUOTA_RESERVATION_PROTOCOL_V1 | QUOTA_RESERVATION_PROTOCOL_V2)) } pub fn check_operation_allowed(&self, current_usage: u64, operation_size: u64) -> bool { @@ -322,9 +344,21 @@ mod tests { } #[test] - fn unknown_reservation_protocol_does_not_activate_v1() { + fn sharded_quota_roundtrips_with_authoritative_reservation_quota() { + let quota = BucketQuota::new_sharded(4096); + let json = serde_json::to_value("a).expect("sharded quota should serialize"); + assert_eq!(json["quota"], 0); + assert_eq!(json["reservation_protocol"], QUOTA_RESERVATION_PROTOCOL_V2); + assert_eq!(json["reservation_quota"], 4096); + let restored: BucketQuota = serde_json::from_value(json).expect("sharded quota should parse"); + assert_eq!(restored.quota, Some(4096)); + assert!(restored.uses_sharded_reservations()); + } + + #[test] + fn unknown_reservation_protocol_does_not_activate_any_known_protocol() { let quota: BucketQuota = - serde_json::from_str(r#"{"quota":0,"quota_type":"Hard","reservation_protocol":2,"reservation_quota":2048}"#) + serde_json::from_str(r#"{"quota":0,"quota_type":"Hard","reservation_protocol":3,"reservation_quota":2048}"#) .expect("future protocol should remain parseable"); assert!(!quota.uses_durable_reservations()); diff --git a/crates/ecstore/src/bucket/quota/reservation.rs b/crates/ecstore/src/bucket/quota/reservation.rs index 0a10b9200..4819ec6b0 100644 --- a/crates/ecstore/src/bucket/quota/reservation.rs +++ b/crates/ecstore/src/bucket/quota/reservation.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use super::QUOTA_RESERVATION_PROTOCOL_V2; use crate::bucket::metadata_sys; use crate::config::com::{CONFIG_PREFIX, read_config_no_lock, save_config_with_opts}; use crate::data_usage::compute_bucket_usage; @@ -21,11 +22,12 @@ use crate::error::{Result, StorageError, is_err_object_not_found, is_err_version use crate::object_api::{ObjectInfo, ObjectOptions, QuotaAdmission}; use crate::set_disk::{SetDisks, get_lock_acquire_timeout}; use crate::storage_api_contracts::namespace::NamespaceLocking; -use crate::storage_api_contracts::object::ObjectOperations; +use crate::storage_api_contracts::{list::ListOperations as _, object::ObjectOperations}; use crate::store::ECStore; use futures::{StreamExt, stream}; use rustfs_lock::NamespaceLockGuard; use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; use std::collections::BTreeMap; use std::sync::Arc; use std::time::Duration; @@ -42,11 +44,17 @@ const EVENT_QUOTA_ADMISSION: &str = "quota_admission"; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_QUOTA: &str = "quota"; +const SHARDED_LEDGER_FORMAT_VERSION: u8 = 1; +const SHARDED_LEDGER_COUNT: u16 = 16; + #[cfg(any(test, feature = "test-util"))] static FAIL_NEXT_LEDGER_SAVE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); // Lock order: caller-held destination object/upload, bucket metadata -// transaction (read), operation reservation, then quota ledger. +// transaction (read), operation reservation, then quota ledger. Protocol v2 +// uses allocator(read) -> shard(write); credit refill releases the operation +// and shard guards before taking allocator(write), and orphan probes never +// hold allocator/shard guards while acquiring an operation lock. // When Object Lock already holds the metadata transaction read lock, reuse // that guard: reacquiring it behind a waiting metadata writer would deadlock. @@ -202,6 +210,206 @@ impl QuotaLedger { } } +#[derive(Debug, Clone, Serialize, Deserialize)] +struct AllocatorGrant { + shard_index: u16, + amount: u64, + #[serde(default)] + initial_usage: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct QuotaAllocatorLedger { + version: u8, + bucket_incarnation: Uuid, + quota_revision_unix_nanos: i128, + quota_limit: u64, + generation: u64, + grants: BTreeMap, +} + +impl QuotaAllocatorLedger { + fn new(bucket_incarnation: Uuid, quota_revision: OffsetDateTime, quota_limit: u64) -> Self { + Self { + version: SHARDED_LEDGER_FORMAT_VERSION, + bucket_incarnation, + quota_revision_unix_nanos: quota_revision.unix_timestamp_nanos(), + quota_limit, + generation: 0, + grants: BTreeMap::new(), + } + } + + fn matches(&self, bucket_incarnation: Uuid, quota_revision: OffsetDateTime, quota_limit: u64) -> bool { + self.bucket_incarnation == bucket_incarnation + && self.quota_revision_unix_nanos == quota_revision.unix_timestamp_nanos() + && self.quota_limit == quota_limit + } + + fn issued_bytes(&self) -> Result { + self.grants.values().try_fold(0_u64, |total, grant| { + total.checked_add(grant.amount).ok_or(StorageError::PartMissingOrCorrupt) + }) + } + + fn grants_for(&self, shard_index: u16) -> BTreeMap { + self.grants + .iter() + .filter_map(|(grant_id, grant)| (grant.shard_index == shard_index).then_some((*grant_id, grant.clone()))) + .collect() + } +} + +#[derive(Debug, Serialize, Deserialize)] +struct QuotaShardLedger { + version: u8, + bucket_incarnation: Uuid, + quota_revision_unix_nanos: i128, + shard_index: u16, + allocator_generation: u64, + accounted_usage: u64, + grants: BTreeMap, + reservations: BTreeMap, + #[serde(default)] + reconcile_required: bool, + #[serde(default)] + reap_cursor: Option, +} + +impl QuotaShardLedger { + fn new(bucket_incarnation: Uuid, quota_revision: OffsetDateTime, shard_index: u16) -> Self { + Self { + version: SHARDED_LEDGER_FORMAT_VERSION, + bucket_incarnation, + quota_revision_unix_nanos: quota_revision.unix_timestamp_nanos(), + shard_index, + allocator_generation: 0, + accounted_usage: 0, + grants: BTreeMap::new(), + reservations: BTreeMap::new(), + reconcile_required: false, + reap_cursor: None, + } + } + + fn matches(&self, bucket_incarnation: Uuid, quota_revision: OffsetDateTime, shard_index: u16) -> bool { + self.bucket_incarnation == bucket_incarnation + && self.quota_revision_unix_nanos == quota_revision.unix_timestamp_nanos() + && self.shard_index == shard_index + } + + fn credit_limit(&self) -> Result { + self.grants.values().try_fold(0_u64, |total, amount| { + total.checked_add(*amount).ok_or(StorageError::PartMissingOrCorrupt) + }) + } + + fn available_credit(&self) -> Result { + self.credit_limit()? + .checked_sub(self.accounted_usage) + .ok_or(StorageError::PartMissingOrCorrupt) + } + + fn adopt_grants(&mut self, allocator: &QuotaAllocatorLedger) -> Result<()> { + let expected = allocator.grants_for(self.shard_index); + for (grant_id, amount) in &self.grants { + if expected.get(grant_id).map(|grant| grant.amount) != Some(*amount) { + return Err(StorageError::PartMissingOrCorrupt); + } + } + if self.grants.is_empty() && self.accounted_usage == 0 { + self.accounted_usage = expected.values().try_fold(0_u64, |total, grant| { + total + .checked_add(grant.initial_usage) + .ok_or(StorageError::PartMissingOrCorrupt) + })?; + } + self.grants = expected + .into_iter() + .map(|(grant_id, grant)| (grant_id, grant.amount)) + .collect(); + self.credit_limit()? + .checked_sub(self.accounted_usage) + .ok_or(StorageError::PartMissingOrCorrupt)?; + self.allocator_generation = allocator.generation; + Ok(()) + } + + fn reserve(&mut self, operation_id: Uuid, reservation: PersistedReservation) -> Result<()> { + self.accounted_usage = self + .accounted_usage + .checked_add(reservation.growth()) + .ok_or(StorageError::PartMissingOrCorrupt)?; + if self.accounted_usage > self.credit_limit()? { + return Err(StorageError::PartMissingOrCorrupt); + } + self.reservations.insert(operation_id, reservation); + Ok(()) + } + + fn commit(&mut self, operation_id: Uuid, expected: &PersistedReservation) -> Result<()> { + let Some(reservation) = self.reservations.remove(&operation_id) else { + return Err(StorageError::PartMissingOrCorrupt); + }; + if !reservation.matches_expected(expected) { + return Err(StorageError::PartMissingOrCorrupt); + } + if reservation.new_size < reservation.old_size { + self.reconcile_required = true; + } + Ok(()) + } + + fn abort(&mut self, operation_id: Uuid, expected: &PersistedReservation) -> Result<()> { + let Some(reservation) = self.reservations.remove(&operation_id) else { + return Ok(()); + }; + if !reservation.matches_expected(expected) { + return Err(StorageError::PartMissingOrCorrupt); + } + self.accounted_usage = self + .accounted_usage + .checked_sub(reservation.growth()) + .ok_or(StorageError::PartMissingOrCorrupt)?; + Ok(()) + } + + fn mark_commit_started(&mut self, operation_id: Uuid, expected: &PersistedReservation) -> Result<()> { + let reservation = self + .reservations + .get_mut(&operation_id) + .ok_or(StorageError::PartMissingOrCorrupt)?; + if !reservation.matches_expected(expected) { + return Err(StorageError::PartMissingOrCorrupt); + } + reservation.commit_started = true; + Ok(()) + } + + fn reap_candidates(&self, now: i64) -> (Vec, Option) { + let aged = self + .reservations + .iter() + .filter(|(_, reservation)| { + reservation.created_at > now || now.saturating_sub(reservation.created_at) >= ORPHAN_MIN_AGE_SECONDS + }) + .map(|(operation_id, _)| *operation_id) + .collect::>(); + let start = self + .reap_cursor + .and_then(|cursor| aged.iter().position(|operation_id| *operation_id > cursor)) + .unwrap_or(0); + let candidates = aged + .iter() + .cycle() + .skip(start) + .take(aged.len().min(MAX_ORPHAN_PROBES_PER_WRITE)) + .copied() + .collect::>(); + (candidates.clone(), candidates.last().copied()) + } +} + pub(crate) struct QuotaContext { store: Option>, bucket: String, @@ -210,6 +418,7 @@ pub(crate) struct QuotaContext { bucket_incarnation: Option, quota_revision: Option, quota_limit: Option, + reservation_protocol: Option, capability_proof: Option, snapshot_admission: Option, legacy_data_movement: bool, @@ -243,6 +452,9 @@ impl QuotaContext { } return Ok(QuotaReservation::unlimited(self.metadata_guard)); } + if self.reservation_protocol == Some(QUOTA_RESERVATION_PROTOCOL_V2) { + return reserve_sharded(self, old_size, new_size).await; + } let store = self.store.ok_or(StorageError::PartMissingOrCorrupt)?; let bucket_incarnation = self.bucket_incarnation.ok_or(StorageError::PartMissingOrCorrupt)?; let quota_revision = self.quota_revision.ok_or(StorageError::PartMissingOrCorrupt)?; @@ -311,7 +523,7 @@ impl QuotaContext { save_ledger_locked(Arc::clone(&store), &ledger_data.ledger_object, &ledger, &ledger_guard).await?; Ok(QuotaReservation { - ledger: Some(ledger_data), + ledger: Some(ReservationLedgerData::Single(ledger_data)), operation_guard: Some(operation_guard), metadata_guard, capability_proof, @@ -319,10 +531,157 @@ impl QuotaContext { }) }) .await - .map_err(|err| StorageError::other(format!("quota ledger reservation task failed: {err}")))? + .map_err(|err| StorageError::other_with_context("quota ledger reservation task failed", err))? } } +async fn reserve_sharded(context: QuotaContext, old_size: u64, new_size: u64) -> Result { + let quota_limit = context.quota_limit.ok_or(StorageError::PartMissingOrCorrupt)?; + let store = context.store.ok_or(StorageError::PartMissingOrCorrupt)?; + let bucket_incarnation = context.bucket_incarnation.ok_or(StorageError::PartMissingOrCorrupt)?; + let quota_revision = context.quota_revision.ok_or(StorageError::PartMissingOrCorrupt)?; + let operation_id = Uuid::new_v4(); + let shard_index = shard_index(&context.object); + let allocator_object = allocator_object(&context.ledger_object, bucket_incarnation, quota_revision); + let shard_object = shard_object(&context.ledger_object, shard_index, bucket_incarnation, quota_revision); + let growth = new_size.saturating_sub(old_size); + let reservation = PersistedReservation { + object: context.object, + old_size, + new_size, + created_at: now_unix(), + pool_index: context.pool_index, + set_index: context.set_index, + commit_started: false, + }; + let metadata_guard = context.metadata_guard; + let capability_proof = context.capability_proof; + let data = ShardedReservationData { + store: Arc::clone(&store), + bucket: context.bucket, + ledger_object: context.ledger_object, + allocator_object, + shard_object, + shard_index, + bucket_incarnation, + quota_revision, + operation_id, + reservation, + }; + let bootstrap = ensure_sharded_allocator(Arc::clone(&store), &data, growth, quota_limit).await?; + #[cfg(test)] + tests::pause_after_allocator_ready(&data.bucket, &data.reservation.object).await; + let operation_lock_object = operation_lock_object(&data.shard_object, operation_id); + let operation_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &operation_lock_object).await?; + let operation_guard = operation_lock.get_write_lock(get_lock_acquire_timeout()).await?; + + for attempt in 0..=2 { + let allocator_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &data.allocator_object).await?; + let allocator_guard = allocator_lock.get_read_lock(get_lock_acquire_timeout()).await?; + if allocator_guard.is_lock_lost() { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "quota_allocator", + bucket: RUSTFS_META_BUCKET.to_string(), + object: data.allocator_object.clone(), + required: 1, + achieved: 0, + }); + } + // Initialization released its lock; a concurrent refill may have added grants. + let allocator = load_current_allocator_locked( + Arc::clone(&store), + &data.allocator_object, + bucket_incarnation, + quota_revision, + quota_limit, + ) + .await?; + let allow_create = shard_may_be_created(&allocator, data.shard_index, bootstrap); + let shard_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &data.shard_object).await?; + let shard_guard = Arc::new(shard_lock.get_write_lock(get_lock_acquire_timeout()).await?); + let mut shard = load_current_shard_locked( + Arc::clone(&store), + &data.shard_object, + bucket_incarnation, + quota_revision, + data.shard_index, + allow_create, + &allocator, + ) + .await?; + let growth = data.reservation.growth(); + if growth == 0 || shard.available_credit()? >= growth { + if operation_guard.is_lock_lost() + || allocator_guard.is_lock_lost() + || shard_guard.is_lock_lost() + || metadata_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) + { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "quota_sharded_reservation", + bucket: data.bucket.clone(), + object: data.shard_object.clone(), + required: 1, + achieved: 0, + }); + } + fence_namespace_mutations(&store, RUSTFS_META_BUCKET, &data.shard_object, None).await?; + if allocator_guard.is_lock_lost() || shard_guard.is_lock_lost() { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "quota_sharded_reservation", + bucket: data.bucket.clone(), + object: data.shard_object.clone(), + required: 1, + achieved: 0, + }); + } + shard.reserve(data.operation_id, data.reservation.clone())?; + save_shard_locked(Arc::clone(&store), &data.shard_object, &shard, &shard_guard).await?; + return Ok(QuotaReservation { + ledger: Some(ReservationLedgerData::Sharded(data)), + operation_guard: Some(operation_guard), + metadata_guard, + capability_proof, + state: ReservationState::Pending, + }); + } + let shard_was_empty = shard.grants.is_empty() && shard.reservations.is_empty() && shard.accounted_usage == 0; + if shard_was_empty { + fence_namespace_mutations(&store, RUSTFS_META_BUCKET, &data.shard_object, None).await?; + save_shard_locked(Arc::clone(&store), &data.shard_object, &shard, &shard_guard).await?; + } + drop(shard_guard); + drop(allocator_guard); + + if attempt == 2 { + return Err(StorageError::QuotaExceeded { + current: quota_limit, + limit: quota_limit, + }); + } + if !shard_was_empty + && attempt < 2 + && reap_sharded_reservations(Arc::clone(&store), &data, bucket_incarnation, quota_revision, quota_limit).await? + { + continue; + } + refill_sharded_credit( + Arc::clone(&store), + &data.allocator_object, + data.shard_index, + growth, + bucket_incarnation, + quota_revision, + quota_limit, + ) + .await?; + } + + Err(StorageError::QuotaExceeded { + current: quota_limit, + limit: quota_limit, + }) +} + #[derive(Clone)] struct LedgerReservationData { store: Arc, @@ -332,8 +691,51 @@ struct LedgerReservationData { reservation: PersistedReservation, } +#[derive(Clone)] +enum ReservationLedgerData { + Single(LedgerReservationData), + Sharded(ShardedReservationData), +} + +#[derive(Clone)] +struct ShardedReservationData { + store: Arc, + bucket: String, + ledger_object: String, + allocator_object: String, + shard_object: String, + shard_index: u16, + bucket_incarnation: Uuid, + quota_revision: OffsetDateTime, + operation_id: Uuid, + reservation: PersistedReservation, +} + +impl ReservationLedgerData { + fn store(&self) -> &Arc { + match self { + Self::Single(data) => &data.store, + Self::Sharded(data) => &data.store, + } + } + + fn bucket(&self) -> &str { + match self { + Self::Single(data) => &data.bucket, + Self::Sharded(data) => &data.bucket, + } + } + + fn error_object(&self) -> &str { + match self { + Self::Single(data) => &data.ledger_object, + Self::Sharded(data) => &data.shard_object, + } + } +} + pub(crate) struct QuotaReservation { - ledger: Option, + ledger: Option, operation_guard: Option, metadata_guard: Option>, capability_proof: Option, @@ -373,7 +775,7 @@ impl QuotaReservation { pub(crate) async fn mark_commit_started(&mut self) -> Result<()> { if !self.capability_proof_matches() { let ledger = self.ledger.as_ref().ok_or(StorageError::PartMissingOrCorrupt)?; - return Err(quota_capability_error(&ledger.bucket, &ledger.ledger_object)); + return Err(quota_capability_error(ledger.bucket(), ledger.error_object())); } if let Some(ledger) = self.ledger.as_ref() { mark_commit_started(ledger).await?; @@ -387,7 +789,7 @@ impl QuotaReservation { let Some(ledger) = self.ledger.as_ref() else { return; }; - crate::store::list_objects::observe_list_objects_mutation(&ledger.store, &ledger.bucket).await; + crate::store::list_objects::observe_list_objects_mutation(ledger.store(), ledger.bucket()).await; match settle(ledger, true).await { Ok(()) => self.ledger = None, Err(err) => log_deferred_settlement(ledger, "commit_deferred", &err), @@ -456,6 +858,7 @@ pub(crate) async fn begin( bucket_incarnation: None, quota_revision: None, quota_limit: None, + reservation_protocol: None, capability_proof: None, snapshot_admission: None, legacy_data_movement: false, @@ -474,6 +877,7 @@ pub(crate) async fn begin( bucket_incarnation: None, quota_revision: None, quota_limit: Some(snapshot_admission.quota_limit()), + reservation_protocol: None, capability_proof: None, snapshot_admission: Some(snapshot_admission), legacy_data_movement: false, @@ -492,6 +896,7 @@ pub(crate) async fn begin( bucket_incarnation: None, quota_revision: None, quota_limit: None, + reservation_protocol: None, capability_proof: None, snapshot_admission: None, legacy_data_movement: false, @@ -530,6 +935,7 @@ pub(crate) async fn begin( None }; let durable_quota_limit = durable_quota.and_then(|quota| quota.quota); + let reservation_protocol = durable_quota.and_then(|quota| quota.reservation_protocol); let snapshot_admission = match quota.as_ref().filter(|quota| !quota.uses_durable_reservations()) { Some(quota) => match (quota.quota, snapshot_admission) { (Some(limit), Some(admission)) if admission.quota_limit() == limit => Some(admission), @@ -572,6 +978,7 @@ pub(crate) async fn begin( bucket_incarnation: Some(bucket_incarnation), quota_revision: Some(quota_revision), quota_limit, + reservation_protocol, capability_proof, snapshot_admission, legacy_data_movement, @@ -619,7 +1026,13 @@ fn logical_object_size(info: &ObjectInfo) -> Result { crate::data_usage::quota_object_size(info) } -async fn mark_commit_started(data: &LedgerReservationData) -> Result<()> { +async fn mark_commit_started(data: &ReservationLedgerData) -> Result<()> { + if let ReservationLedgerData::Sharded(data) = data { + return mark_commit_started_sharded(data).await; + } + let ReservationLedgerData::Single(data) = data else { + unreachable!(); + }; let data = data.clone(); tokio::spawn(async move { let ledger_lock = data.store.new_ns_lock(RUSTFS_META_BUCKET, &data.ledger_object).await?; @@ -630,10 +1043,16 @@ async fn mark_commit_started(data: &LedgerReservationData) -> Result<()> { save_ledger_locked(Arc::clone(&data.store), &data.ledger_object, &ledger, &ledger_guard).await }) .await - .map_err(|err| StorageError::other(format!("quota commit marker task failed: {err}")))? + .map_err(|err| StorageError::other_with_context("quota commit marker task failed", err))? } -async fn settle(data: &LedgerReservationData, committed: bool) -> Result<()> { +async fn settle(data: &ReservationLedgerData, committed: bool) -> Result<()> { + if let ReservationLedgerData::Sharded(data) = data { + return settle_sharded(data, committed).await; + } + let ReservationLedgerData::Single(data) = data else { + unreachable!(); + }; let store = Arc::clone(&data.store); let ledger_object = data.ledger_object.clone(); let operation_id = data.operation_id; @@ -655,7 +1074,7 @@ async fn settle(data: &LedgerReservationData, committed: bool) -> Result<()> { save_ledger_locked(Arc::clone(&store), &ledger_object, &ledger, &ledger_guard).await }) .await - .map_err(|err| StorageError::other(format!("quota ledger settlement task failed: {err}")))? + .map_err(|err| StorageError::other_with_context("quota ledger settlement task failed", err))? } async fn load_current_ledger_locked( @@ -789,6 +1208,51 @@ async fn exact_bucket_usage(store: &Arc, bucket: &str) -> Result { Ok(compute_bucket_usage(Arc::clone(store), bucket).await?.size) } +async fn reconcile_shard_exact(store: &Arc, bucket: &str, shard: &mut QuotaShardLedger) -> Result<()> { + if !shard.reservations.is_empty() { + return Err(StorageError::PartMissingOrCorrupt); + } + shard.accounted_usage = exact_shard_usage(store, bucket, shard.shard_index).await?; + if shard.accounted_usage > shard.credit_limit()? { + return Err(StorageError::PartMissingOrCorrupt); + } + shard.reconcile_required = false; + Ok(()) +} + +async fn exact_shard_usage(store: &Arc, bucket: &str, target_shard: u16) -> Result { + let usages = exact_shard_usages(store, bucket).await?; + usages + .get(usize::from(target_shard)) + .copied() + .ok_or(StorageError::PartMissingOrCorrupt) +} + +async fn exact_shard_usages(store: &Arc, bucket: &str) -> Result> { + let mut marker = None; + let mut version_marker = None; + let mut usages = vec![0_u64; usize::from(SHARDED_LEDGER_COUNT)]; + loop { + let page = Arc::clone(store) + .list_object_versions(bucket, "", marker, version_marker, None, 1_000) + .await?; + for object in page.objects { + let shard = usize::from(shard_index(object.name.as_str())); + usages[shard] = usages[shard] + .checked_add(logical_object_size(&object)?) + .ok_or(StorageError::PartMissingOrCorrupt)?; + } + if !page.is_truncated { + return Ok(usages); + } + marker = page.next_marker; + version_marker = page.next_version_idmarker; + if marker.is_none() && version_marker.is_none() { + return Err(StorageError::PartMissingOrCorrupt); + } + } +} + fn ledger_object(bucket: &str) -> String { format!("{CONFIG_PREFIX}/quota-ledger/{bucket}.json") } @@ -797,6 +1261,512 @@ fn operation_lock_object(ledger_object: &str, operation_id: Uuid) -> String { format!("{ledger_object}.operations/{operation_id}") } +fn allocator_object(ledger_object: &str, bucket_incarnation: Uuid, quota_revision: OffsetDateTime) -> String { + format!( + "{ledger_object}.allocator.{bucket_incarnation}.{}.json", + quota_revision.unix_timestamp_nanos() + ) +} + +fn shard_object(ledger_object: &str, shard_index: u16, bucket_incarnation: Uuid, quota_revision: OffsetDateTime) -> String { + format!( + "{ledger_object}.shards/{bucket_incarnation}/{}.{shard_index:02}.json", + quota_revision.unix_timestamp_nanos() + ) +} + +fn shard_index(object: &str) -> u16 { + let digest = Sha256::digest(object.as_bytes()); + u16::from_be_bytes([digest[0], digest[1]]) % SHARDED_LEDGER_COUNT +} + +fn shard_may_be_created(allocator: &QuotaAllocatorLedger, shard_index: u16, bootstrap: bool) -> bool { + if bootstrap { + return true; + } + let grants = allocator.grants_for(shard_index); + grants.is_empty() || grants.values().all(|grant| grant.initial_usage > 0) +} + +fn credit_grant_amount(issued: u64, quota_limit: u64, growth: u64) -> Result { + let available = quota_limit.saturating_sub(issued); + if growth > available { + return Err(StorageError::QuotaExceeded { + current: issued, + limit: quota_limit, + }); + } + let fair_chunk = quota_limit.div_ceil(SHARDED_LEDGER_COUNT as u64).max(1); + Ok(growth.max(fair_chunk).min(available)) +} + +async fn ensure_sharded_allocator( + store: Arc, + data: &ShardedReservationData, + growth: u64, + quota_limit: u64, +) -> Result { + let allocator_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &data.allocator_object).await?; + let allocator_guard = allocator_lock.get_read_lock(get_lock_acquire_timeout()).await?; + match load_allocator_locked(Arc::clone(&store), &data.allocator_object).await { + Ok(allocator) if allocator.matches(data.bucket_incarnation, data.quota_revision, quota_limit) => { + return Ok(false); + } + Ok(_) | Err(StorageError::ConfigNotFound) => {} + Err(err) => return Err(err), + } + drop(allocator_guard); + + let allocator_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &data.allocator_object).await?; + let allocator_guard = Arc::new(allocator_lock.get_write_lock(get_lock_acquire_timeout()).await?); + match load_allocator_locked(Arc::clone(&store), &data.allocator_object).await { + Ok(allocator) if allocator.matches(data.bucket_incarnation, data.quota_revision, quota_limit) => { + return Ok(false); + } + Ok(allocator) if !allocator.grants.is_empty() => return Err(StorageError::PartMissingOrCorrupt), + Ok(_) | Err(StorageError::ConfigNotFound) => {} + Err(err) => return Err(err), + } + for candidate in 0..SHARDED_LEDGER_COUNT { + let object = shard_object(&data.ledger_object, candidate, data.bucket_incarnation, data.quota_revision); + match read_config_no_lock(Arc::clone(&store), &object).await { + Ok(_) => return Err(StorageError::PartMissingOrCorrupt), + Err(StorageError::ConfigNotFound) => {} + Err(err) => return Err(err), + } + } + if allocator_guard.is_lock_lost() { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "quota_allocator_bootstrap", + bucket: RUSTFS_META_BUCKET.to_string(), + object: data.allocator_object.clone(), + required: 1, + achieved: 0, + }); + } + fence_namespace_mutations(&store, RUSTFS_META_BUCKET, &data.allocator_object, None).await?; + let shard_usages = exact_shard_usages(&store, &data.bucket).await?; + let usage = shard_usages + .iter() + .try_fold(0_u64, |total, usage| total.checked_add(*usage).ok_or(StorageError::PartMissingOrCorrupt))?; + let required = usage.checked_add(growth).ok_or(StorageError::PartMissingOrCorrupt)?; + let _ = credit_grant_amount(0, quota_limit, required)?; + let mut allocator = QuotaAllocatorLedger::new(data.bucket_incarnation, data.quota_revision, quota_limit); + for (shard_index, initial_usage) in shard_usages.into_iter().enumerate() { + let shard_index = u16::try_from(shard_index).map_err(|_| StorageError::PartMissingOrCorrupt)?; + let growth_for_shard = if shard_index == data.shard_index { growth } else { 0 }; + let amount = initial_usage + .checked_add(growth_for_shard) + .ok_or(StorageError::PartMissingOrCorrupt)?; + if amount == 0 { + continue; + } + allocator.grants.insert( + Uuid::new_v4(), + AllocatorGrant { + shard_index, + amount, + initial_usage, + }, + ); + } + allocator.generation = 1; + save_allocator_locked(Arc::clone(&store), &data.allocator_object, &allocator, &allocator_guard).await?; + Ok(true) +} + +async fn load_allocator_locked(store: Arc, object: &str) -> Result { + let data = read_config_no_lock(store, object).await?; + let allocator: QuotaAllocatorLedger = serde_json::from_slice(&data)?; + if allocator.version != SHARDED_LEDGER_FORMAT_VERSION { + return Err(StorageError::CorruptedFormat); + } + allocator.issued_bytes()?; + Ok(allocator) +} + +async fn load_current_allocator_locked( + store: Arc, + object: &str, + bucket_incarnation: Uuid, + quota_revision: OffsetDateTime, + quota_limit: u64, +) -> Result { + match load_allocator_locked(Arc::clone(&store), object).await { + Ok(allocator) if allocator.matches(bucket_incarnation, quota_revision, quota_limit) => Ok(allocator), + Ok(allocator) if allocator.grants.is_empty() => { + Ok(QuotaAllocatorLedger::new(bucket_incarnation, quota_revision, quota_limit)) + } + Ok(_) => Err(StorageError::PartMissingOrCorrupt), + Err(StorageError::ConfigNotFound) => Err(StorageError::PartMissingOrCorrupt), + Err(err) => Err(err), + } +} + +async fn load_shard_locked(store: Arc, object: &str) -> Result { + let data = read_config_no_lock(store, object).await?; + let shard: QuotaShardLedger = serde_json::from_slice(&data)?; + if shard.version != SHARDED_LEDGER_FORMAT_VERSION { + return Err(StorageError::CorruptedFormat); + } + shard + .credit_limit()? + .checked_sub(shard.accounted_usage) + .ok_or(StorageError::PartMissingOrCorrupt)?; + Ok(shard) +} + +async fn load_current_shard_locked( + store: Arc, + object: &str, + bucket_incarnation: Uuid, + quota_revision: OffsetDateTime, + shard_index: u16, + allow_create: bool, + allocator: &QuotaAllocatorLedger, +) -> Result { + let mut shard = match load_shard_locked(Arc::clone(&store), object).await { + Ok(shard) if shard.matches(bucket_incarnation, quota_revision, shard_index) => shard, + Ok(_) => return Err(StorageError::PartMissingOrCorrupt), + Err(StorageError::ConfigNotFound) if allow_create => { + QuotaShardLedger::new(bucket_incarnation, quota_revision, shard_index) + } + Err(StorageError::ConfigNotFound) => return Err(StorageError::PartMissingOrCorrupt), + Err(err) => return Err(err), + }; + shard.adopt_grants(allocator)?; + Ok(shard) +} + +async fn save_allocator_locked( + store: Arc, + object: &str, + allocator: &QuotaAllocatorLedger, + guard: &Arc, +) -> Result<()> { + if guard.is_lock_lost() { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "quota_allocator", + bucket: RUSTFS_META_BUCKET.to_string(), + object: object.to_string(), + required: 1, + achieved: 0, + }); + } + let mut opts = ObjectOptions { + max_parity: true, + no_lock: true, + ..Default::default() + }; + let _ = opts.set_quota_admission(0, u64::MAX); + opts.add_owned_write_lock(Arc::clone(guard), RUSTFS_META_BUCKET, object); + opts.write_completion = crate::object_api::WriteCompletion::TailDrained; + save_config_with_opts(store, object, serde_json::to_vec(allocator)?, &opts).await +} + +async fn save_shard_locked( + store: Arc, + object: &str, + shard: &QuotaShardLedger, + guard: &Arc, +) -> Result<()> { + if guard.is_lock_lost() { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "quota_shard", + bucket: RUSTFS_META_BUCKET.to_string(), + object: object.to_string(), + required: 1, + achieved: 0, + }); + } + let mut opts = ObjectOptions { + max_parity: true, + no_lock: true, + ..Default::default() + }; + let _ = opts.set_quota_admission(0, u64::MAX); + opts.add_owned_write_lock(Arc::clone(guard), RUSTFS_META_BUCKET, object); + opts.write_completion = crate::object_api::WriteCompletion::TailDrained; + save_config_with_opts(store, object, serde_json::to_vec(shard)?, &opts).await +} + +async fn refill_sharded_credit( + store: Arc, + allocator_object: &str, + shard_index: u16, + growth: u64, + bucket_incarnation: Uuid, + quota_revision: OffsetDateTime, + quota_limit: u64, +) -> Result<()> { + let allocator_lock = store.new_ns_lock(RUSTFS_META_BUCKET, allocator_object).await?; + let allocator_guard = Arc::new(allocator_lock.get_write_lock(get_lock_acquire_timeout()).await?); + if allocator_guard.is_lock_lost() { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "quota_allocator", + bucket: RUSTFS_META_BUCKET.to_string(), + object: allocator_object.to_string(), + required: 1, + achieved: 0, + }); + } + fence_namespace_mutations(&store, RUSTFS_META_BUCKET, allocator_object, None).await?; + let mut allocator = + load_current_allocator_locked(Arc::clone(&store), allocator_object, bucket_incarnation, quota_revision, quota_limit) + .await?; + let issued = allocator.issued_bytes()?; + let amount = credit_grant_amount(issued, quota_limit, growth)?; + let grant_id = Uuid::new_v4(); + allocator.grants.insert( + grant_id, + AllocatorGrant { + shard_index, + amount, + initial_usage: 0, + }, + ); + allocator.generation = allocator + .generation + .checked_add(1) + .ok_or(StorageError::PartMissingOrCorrupt)?; + save_allocator_locked(Arc::clone(&store), allocator_object, &allocator, &allocator_guard).await +} + +async fn reap_sharded_reservations( + store: Arc, + data: &ShardedReservationData, + bucket_incarnation: Uuid, + quota_revision: OffsetDateTime, + quota_limit: u64, +) -> Result { + let allocator_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &data.allocator_object).await?; + let allocator_guard = allocator_lock.get_read_lock(get_lock_acquire_timeout()).await?; + if allocator_guard.is_lock_lost() { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "quota_allocator", + bucket: RUSTFS_META_BUCKET.to_string(), + object: data.allocator_object.clone(), + required: 1, + achieved: 0, + }); + } + let allocator = load_current_allocator_locked( + Arc::clone(&store), + &data.allocator_object, + bucket_incarnation, + quota_revision, + quota_limit, + ) + .await?; + let shard_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &data.shard_object).await?; + let shard_guard = shard_lock.get_write_lock(get_lock_acquire_timeout()).await?; + let shard = load_current_shard_locked( + Arc::clone(&store), + &data.shard_object, + bucket_incarnation, + quota_revision, + data.shard_index, + false, + &allocator, + ) + .await?; + let (candidates, next_cursor) = shard.reap_candidates(now_unix()); + drop(shard_guard); + drop(allocator_guard); + if candidates.is_empty() && next_cursor.is_none() { + return Ok(false); + } + let probe_results = stream::iter(candidates) + .map(|operation_id| { + let store = Arc::clone(&store); + let object = data.shard_object.clone(); + async move { + let lock_object = operation_lock_object(&object, operation_id); + let operation_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &lock_object).await?; + Ok::<_, StorageError>( + operation_lock + .get_write_lock_quiet(Duration::from_millis(50)) + .await + .ok() + .map(|guard| (operation_id, guard)), + ) + } + }) + .buffer_unordered(ORPHAN_PROBE_CONCURRENCY) + .collect::>() + .await; + let mut orphan_guards = Vec::new(); + for result in probe_results { + if let Some(guard) = result? { + orphan_guards.push(guard); + } + } + let had_orphan_guards = !orphan_guards.is_empty(); + let allocator_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &data.allocator_object).await?; + let allocator_guard = allocator_lock.get_read_lock(get_lock_acquire_timeout()).await?; + if allocator_guard.is_lock_lost() { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "quota_allocator", + bucket: RUSTFS_META_BUCKET.to_string(), + object: data.allocator_object.clone(), + required: 1, + achieved: 0, + }); + } + let allocator = load_current_allocator_locked( + Arc::clone(&store), + &data.allocator_object, + bucket_incarnation, + quota_revision, + quota_limit, + ) + .await?; + let shard_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &data.shard_object).await?; + let shard_guard = Arc::new(shard_lock.get_write_lock(get_lock_acquire_timeout()).await?); + let mut shard = load_current_shard_locked( + Arc::clone(&store), + &data.shard_object, + bucket_incarnation, + quota_revision, + data.shard_index, + false, + &allocator, + ) + .await?; + let mut removed = false; + for (operation_id, _guard) in &orphan_guards { + let Some(reservation) = shard.reservations.get(operation_id).cloned() else { + continue; + }; + if reservation.commit_started { + shard.reservations.remove(operation_id); + shard.reconcile_required = true; + } else { + shard.abort(*operation_id, &reservation)?; + } + removed = true; + } + let cursor_changed = next_cursor.is_some() && shard.reap_cursor != next_cursor; + if next_cursor.is_some() { + shard.reap_cursor = next_cursor; + } + if !removed && !cursor_changed { + return Ok(false); + } + if shard.reservations.is_empty() && shard.reconcile_required { + reconcile_shard_exact(&store, &data.bucket, &mut shard).await?; + } + fence_namespace_mutations(&store, RUSTFS_META_BUCKET, &data.shard_object, None).await?; + if allocator_guard.is_lock_lost() || shard_guard.is_lock_lost() { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "quota_shard_reap", + bucket: data.bucket.clone(), + object: data.shard_object.clone(), + required: 1, + achieved: 0, + }); + } + save_shard_locked(Arc::clone(&store), &data.shard_object, &shard, &shard_guard).await?; + Ok(removed || cursor_changed || had_orphan_guards) +} + +async fn mark_commit_started_sharded(data: &ShardedReservationData) -> Result<()> { + let store = Arc::clone(&data.store); + let data = data.clone(); + tokio::spawn(async move { + let allocator_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &data.allocator_object).await?; + let allocator_guard = allocator_lock.get_read_lock(get_lock_acquire_timeout()).await?; + if allocator_guard.is_lock_lost() { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "quota_allocator", + bucket: RUSTFS_META_BUCKET.to_string(), + object: data.allocator_object.clone(), + required: 1, + achieved: 0, + }); + } + let allocator = load_allocator_locked(Arc::clone(&store), &data.allocator_object).await?; + let shard_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &data.shard_object).await?; + let shard_guard = Arc::new(shard_lock.get_write_lock(get_lock_acquire_timeout()).await?); + let mut shard = load_current_shard_locked( + Arc::clone(&store), + &data.shard_object, + data.bucket_incarnation, + data.quota_revision, + data.shard_index, + false, + &allocator, + ) + .await?; + fence_namespace_mutations(&store, RUSTFS_META_BUCKET, &data.shard_object, None).await?; + if allocator_guard.is_lock_lost() || shard_guard.is_lock_lost() { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "quota_sharded_commit_marker", + bucket: data.bucket.clone(), + object: data.shard_object.clone(), + required: 1, + achieved: 0, + }); + } + shard.mark_commit_started(data.operation_id, &data.reservation)?; + save_shard_locked(Arc::clone(&store), &data.shard_object, &shard, &shard_guard).await + }) + .await + .map_err(|err| StorageError::other_with_context("quota sharded commit marker task failed", err))? +} + +async fn settle_sharded(data: &ShardedReservationData, committed: bool) -> Result<()> { + let store = Arc::clone(&data.store); + let data = data.clone(); + tokio::spawn(async move { + let allocator_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &data.allocator_object).await?; + let allocator_guard = allocator_lock.get_read_lock(get_lock_acquire_timeout()).await?; + if allocator_guard.is_lock_lost() { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "quota_allocator", + bucket: RUSTFS_META_BUCKET.to_string(), + object: data.allocator_object.clone(), + required: 1, + achieved: 0, + }); + } + let allocator = load_allocator_locked(Arc::clone(&store), &data.allocator_object).await?; + let shard_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &data.shard_object).await?; + let shard_guard = Arc::new(shard_lock.get_write_lock(get_lock_acquire_timeout()).await?); + let mut shard = load_current_shard_locked( + Arc::clone(&store), + &data.shard_object, + data.bucket_incarnation, + data.quota_revision, + data.shard_index, + false, + &allocator, + ) + .await?; + fence_namespace_mutations(&store, RUSTFS_META_BUCKET, &data.shard_object, None).await?; + if allocator_guard.is_lock_lost() || shard_guard.is_lock_lost() { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "quota_sharded_settlement", + bucket: data.bucket.clone(), + object: data.shard_object.clone(), + required: 1, + achieved: 0, + }); + } + if committed { + shard.commit(data.operation_id, &data.reservation)?; + } else { + shard.abort(data.operation_id, &data.reservation)?; + } + if shard.reservations.is_empty() && shard.reconcile_required { + reconcile_shard_exact(&store, &data.bucket, &mut shard).await?; + } + save_shard_locked(Arc::clone(&store), &data.shard_object, &shard, &shard_guard).await + }) + .await + .map_err(|err| StorageError::other_with_context("quota sharded settlement task failed", err))? +} + fn now_unix() -> i64 { OffsetDateTime::now_utc().unix_timestamp() } @@ -931,14 +1901,18 @@ fn log_admission_rejected(bucket: &str, object: &str, state: &'static str) { ); } -fn log_deferred_settlement(data: &LedgerReservationData, state: &'static str, err: &StorageError) { +fn log_deferred_settlement(data: &ReservationLedgerData, state: &'static str, err: &StorageError) { + let operation_id = match data { + ReservationLedgerData::Single(data) => data.operation_id, + ReservationLedgerData::Sharded(data) => data.operation_id, + }; warn!( event = EVENT_QUOTA_LEDGER_SETTLEMENT, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_QUOTA, state, - bucket = %data.bucket, - operation_id = %data.operation_id, + bucket = %data.bucket(), + operation_id = %operation_id, error = %err, "quota ledger settlement deferred" ); @@ -1138,4 +2112,275 @@ mod tests { assert!(should_settle_on_drop(ReservationState::Pending)); assert!(should_settle_on_drop(ReservationState::Committed)); } + + #[test] + fn allocator_never_issues_credit_above_the_hard_limit() { + let revision = OffsetDateTime::now_utc(); + let mut allocator = QuotaAllocatorLedger::new(Uuid::new_v4(), revision, 100); + allocator.grants.insert( + Uuid::new_v4(), + AllocatorGrant { + shard_index: 0, + amount: 64, + initial_usage: 0, + }, + ); + assert_eq!(allocator.issued_bytes().expect("issued bytes should sum"), 64); + assert_eq!(credit_grant_amount(64, allocator.quota_limit, 36).expect("exact remaining credit"), 36); + assert!(matches!( + credit_grant_amount(64, allocator.quota_limit, 37), + Err(StorageError::QuotaExceeded { .. }) + )); + } + + #[test] + fn shard_adopts_a_durable_grant_after_allocator_crash_window() { + let revision = OffsetDateTime::now_utc(); + let incarnation = Uuid::new_v4(); + let grant_id = Uuid::new_v4(); + let mut allocator = QuotaAllocatorLedger::new(incarnation, revision, 1024); + allocator.grants.insert( + grant_id, + AllocatorGrant { + shard_index: 3, + amount: 128, + initial_usage: 0, + }, + ); + allocator.generation = 7; + let mut shard = QuotaShardLedger::new(incarnation, revision, 3); + shard.adopt_grants(&allocator).expect("shard should recover the issued grant"); + assert_eq!(shard.grants.get(&grant_id), Some(&128)); + assert_eq!(shard.allocator_generation, 7); + + let operation_id = Uuid::new_v4(); + let reservation = PersistedReservation { + object: "object".to_string(), + old_size: 0, + new_size: 64, + created_at: 0, + pool_index: None, + set_index: None, + commit_started: false, + }; + shard + .reserve(operation_id, reservation.clone()) + .expect("credit should admit reservation"); + assert_eq!(shard.available_credit().expect("available credit"), 64); + shard + .abort(operation_id, &reservation) + .expect("abort should release shard credit"); + assert_eq!(shard.available_credit().expect("available credit"), 128); + } + + #[test] + fn shard_bootstrap_usage_stays_partitioned_by_object_shard() { + let revision = OffsetDateTime::now_utc(); + let incarnation = Uuid::new_v4(); + let mut allocator = QuotaAllocatorLedger::new(incarnation, revision, 100_000); + allocator.grants.insert( + Uuid::new_v4(), + AllocatorGrant { + shard_index: 1, + amount: 60_000, + initial_usage: 60_000, + }, + ); + allocator.grants.insert( + Uuid::new_v4(), + AllocatorGrant { + shard_index: 7, + amount: 30_000, + initial_usage: 30_000, + }, + ); + + let mut first = QuotaShardLedger::new(incarnation, revision, 1); + first.adopt_grants(&allocator).expect("first shard should adopt its grant"); + let mut second = QuotaShardLedger::new(incarnation, revision, 7); + second.adopt_grants(&allocator).expect("second shard should adopt its grant"); + + assert_eq!(first.accounted_usage, 60_000); + assert_eq!(second.accounted_usage, 30_000); + assert_eq!(first.available_credit().expect("first shard credit"), 0); + assert_eq!(second.available_credit().expect("second shard credit"), 0); + } + + #[test] + fn untouched_bootstrap_shard_remains_creatable_after_allocator_refill() { + let revision = OffsetDateTime::now_utc(); + let incarnation = Uuid::new_v4(); + let mut allocator = QuotaAllocatorLedger::new(incarnation, revision, 100_000); + allocator.generation = 2; + allocator.grants.insert( + Uuid::new_v4(), + AllocatorGrant { + shard_index: 7, + amount: 100, + initial_usage: 100, + }, + ); + assert!(shard_may_be_created(&allocator, 7, false)); + + allocator.grants.insert( + Uuid::new_v4(), + AllocatorGrant { + shard_index: 7, + amount: 100, + initial_usage: 0, + }, + ); + assert!(!shard_may_be_created(&allocator, 7, false)); + assert!(shard_may_be_created(&allocator, 1, false)); + } + + #[test] + fn shard_rejects_a_grant_not_owned_by_the_allocator_epoch() { + let revision = OffsetDateTime::now_utc(); + let incarnation = Uuid::new_v4(); + let mut allocator = QuotaAllocatorLedger::new(incarnation, revision, 1024); + allocator.generation = 2; + let mut shard = QuotaShardLedger::new(incarnation, revision, 1); + shard.grants.insert(Uuid::new_v4(), 64); + assert!(matches!(shard.adopt_grants(&allocator), Err(StorageError::PartMissingOrCorrupt))); + } + + #[test] + fn shard_commit_started_shrink_requires_exact_reconciliation() { + let revision = OffsetDateTime::now_utc(); + let incarnation = Uuid::new_v4(); + let mut allocator = QuotaAllocatorLedger::new(incarnation, revision, 256); + let grant_id = Uuid::new_v4(); + allocator.grants.insert( + grant_id, + AllocatorGrant { + shard_index: 0, + amount: 256, + initial_usage: 128, + }, + ); + let mut shard = QuotaShardLedger::new(incarnation, revision, 0); + shard.adopt_grants(&allocator).expect("grant should be adopted"); + let operation_id = Uuid::new_v4(); + let reservation = PersistedReservation { + object: "object".to_string(), + old_size: 128, + new_size: 1, + created_at: 0, + pool_index: None, + set_index: None, + commit_started: true, + }; + shard.reservations.insert(operation_id, reservation.clone()); + shard.commit(operation_id, &reservation).expect("commit marker should settle"); + assert!(shard.reconcile_required); + } + + struct AllocatorPause { + bucket: String, + object: String, + reached: tokio::sync::Notify, + release: tokio::sync::Notify, + } + + static ALLOCATOR_PAUSE: std::sync::Mutex>> = std::sync::Mutex::new(None); + + struct AllocatorPauseGuard(Arc); + + impl Drop for AllocatorPauseGuard { + fn drop(&mut self) { + self.0.release.notify_one(); + let mut pause = ALLOCATOR_PAUSE.lock().expect("allocator pause registry"); + if pause.as_ref().is_some_and(|current| Arc::ptr_eq(current, &self.0)) { + *pause = None; + } + } + } + + pub(super) async fn pause_after_allocator_ready(bucket: &str, object: &str) { + let pause = ALLOCATOR_PAUSE + .lock() + .expect("allocator pause registry") + .as_ref() + .filter(|pause| pause.bucket == bucket && pause.object == object) + .cloned(); + if let Some(pause) = pause { + pause.reached.notify_one(); + pause.release.notified().await; + } + } + + #[tokio::test] + async fn sharded_reservation_reloads_allocator_after_concurrent_refill() { + use crate::object_api::PutObjReader; + use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions}; + use crate::storage_api_contracts::object::ObjectIO; + + let (_directories, store) = metadata_sys::test_support::isolated_store_over_temp_disks().await; + metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await; + crate::services::notification_sys::install_cross_pool_fence_fleet_proof_for_test(); + let bucket = format!("quota-refill-{}", Uuid::new_v4()); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create quota bucket"); + store + .update_bucket_metadata_config( + &bucket, + "quota.json", + serde_json::to_vec(&super::super::BucketQuota::new_sharded(100_000)).expect("serialize v2 quota"), + ) + .await + .expect("enable v2 quota"); + let names = (0..1000) + .map(|index| format!("quota-object-{index}")) + .filter(|name| shard_index(name) == 0) + .take(3) + .collect::>(); + assert_eq!(names.len(), 3); + let mut seed = PutObjReader::from_vec(vec![1]); + store + .put_object(&bucket, &names[0], &mut seed, &ObjectOptions::default()) + .await + .expect("seed a one-byte initial grant"); + + let pause = Arc::new(AllocatorPause { + bucket: bucket.clone(), + object: names[1].clone(), + reached: tokio::sync::Notify::new(), + release: tokio::sync::Notify::new(), + }); + let _pause_guard = AllocatorPauseGuard(Arc::clone(&pause)); + *ALLOCATOR_PAUSE.lock().expect("install allocator pause") = Some(Arc::clone(&pause)); + let waiting_put = tokio::spawn({ + let store = Arc::clone(&store); + let bucket = bucket.clone(); + let object = names[1].clone(); + async move { + let mut body = PutObjReader::from_vec(vec![2]); + store.put_object(&bucket, &object, &mut body, &ObjectOptions::default()).await + } + }); + tokio::time::timeout(Duration::from_secs(30), pause.reached.notified()) + .await + .expect("first PUT reached the gap after allocator initialization"); + + // This PUT refills the allocator and persists the new grant in the same shard. + let mut body = PutObjReader::from_vec(vec![3, 4]); + let refill_result = store + .put_object(&bucket, &names[2], &mut body, &ObjectOptions::default()) + .await; + pause.release.notify_one(); + let waiting_result = tokio::time::timeout(Duration::from_secs(30), waiting_put) + .await + .expect("paused PUT completed") + .expect("paused PUT task joined"); + assert_eq!(refill_result.expect("refill PUT remains below quota").size, 2); + assert_eq!( + waiting_result + .expect("paused PUT must adopt the current allocator grants") + .size, + 1 + ); + } } diff --git a/crates/scanner/src/scanner/tests/quota_reset_preservation.rs b/crates/scanner/src/scanner/tests/quota_reset_preservation.rs index a1de2efc4..66236921f 100644 --- a/crates/scanner/src/scanner/tests/quota_reset_preservation.rs +++ b/crates/scanner/src/scanner/tests/quota_reset_preservation.rs @@ -96,7 +96,7 @@ async fn quota_reset_preservation_unknown_protocol_rejects_put_after_restart() { let (_directory, store, incarnation, path, bytes) = reservation_fixture().await; let mut quota = serde_json::json!({ "quota_type": "Hard", - "reservation_protocol": 2, + "reservation_protocol": 3, "reservation_quota": 1024 }); match quota_shape { diff --git a/docs/operations/durable-quota-reservation-v2.md b/docs/operations/durable-quota-reservation-v2.md new file mode 100644 index 000000000..82200dcd3 --- /dev/null +++ b/docs/operations/durable-quota-reservation-v2.md @@ -0,0 +1,79 @@ +# Durable quota reservation protocol v2 + +Protocol v2 is an opt-in hard-quota protocol for distributed deployments. It +keeps the quota limit authoritative in one allocator ledger while moving +request reservations into sixteen independently locked shard ledgers. + +## State and invariants + +The allocator ledger contains the bucket incarnation, quota revision, limit, +monotonic generation, and durable credit grants. A grant belongs to exactly +one shard and is never larger than the remaining quota at the allocator write +that issued it. A shard ledger contains the same incarnation/revision, the +allocator generation it adopted, its grants, accounted usage, and pending +reservations. + +The safety invariant is: + +```text +sum(shard grant amounts) <= quota_limit +shard.accounted_usage <= sum(shard grant amounts) +``` + +`accounted_usage` includes committed bytes and pending positive reservation +growth. An abort removes the pending growth; a commit removes only the pending +record because its growth is already accounted. A reservation whose commit +marker was persisted is removed only as a precharged reconcile candidate; the +shard is then rebuilt from objects belonging to that shard. + +## Locking and fencing + +The v2 lock order is: + +```text +caller object/upload -> metadata transaction -> operation lock -> allocator +read lock -> shard write lock +``` + +Credit refill releases the operation and shard locks before taking the +allocator write lock. The allocator write is fenced before persistence. Shard +reservation and settlement fence the shard before persistence. The allocator +generation and bucket incarnation/revision are checked on every load, so a +late writer from an older epoch fails closed. Allocator and shard paths include +the bucket incarnation and quota revision, so a quota change starts a new epoch +after an exact bucket bootstrap instead of reusing old grants. + +## Crash and replay behavior + +Allocator grants are durable before a shard can adopt them. If a process exits +after the allocator write but before the shard write, the next shard writer +adopts the grant by ID. If a process exits after a shard reservation, the +operation lock probe can reclaim reservations and marks `commit_started` +entries for exact shard reconciliation, preventing a crash window from +releasing credit for an object that may already be durable. +Repeated commit/abort operations are keyed by the operation ID and expected +reservation fields; mismatches fail closed. + +## Mixed-version rollout + +`reservation_protocol: 2` is serialized with `reservation_quota`. A node that +does not understand v2 must reject the configuration rather than treating it +as the legacy snapshot protocol. Operators must upgrade the whole fleet before +calling `BucketQuota::new_sharded`. Downgrades require clearing or reconciling +all v2 allocator/shard ledgers first; v1 and v2 ledgers are never mixed. + +The default `BucketQuota::new` remains protocol v1. This keeps existing +deployments compatible until an operator has completed the rollout and chosen +the sharded protocol explicitly. + +## Performance boundary + +Normal v2 requests take a shared allocator read lock and one shard write lock; +the bucket-wide allocator write lock is used only to issue a new credit chunk. +The per-grant chunk is bounded by roughly one sixteenth of the quota (or the +requested growth when larger), so one active shard cannot consume the entire +remaining quota through a single speculative grant. It is a conservative upper +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. diff --git a/rustfs/src/admin/handlers/bucket_meta.rs b/rustfs/src/admin/handlers/bucket_meta.rs index e6a8ef9bb..1a23d64dd 100644 --- a/rustfs/src/admin/handlers/bucket_meta.rs +++ b/rustfs/src/admin/handlers/bucket_meta.rs @@ -1369,7 +1369,7 @@ mod import_persist_tests { let unknown_protocol = vec![( format!("bucket/{BUCKET_QUOTA_CONFIG_FILE}"), - br#"{"quota":0,"reservation_protocol":2,"reservation_quota":1024}"#.to_vec(), + br#"{"quota":0,"reservation_protocol":3,"reservation_quota":1024}"#.to_vec(), )]; assert!(imported_quota_requires_fleet_proof(&unknown_protocol).is_err()); } diff --git a/rustfs/src/admin/handlers/quota.rs b/rustfs/src/admin/handlers/quota.rs index 8f7f672e9..8c6f2e18b 100644 --- a/rustfs/src/admin/handlers/quota.rs +++ b/rustfs/src/admin/handlers/quota.rs @@ -46,6 +46,9 @@ pub struct SetBucketQuotaRequest { pub quota: Option, #[serde(default = "default_quota_type")] pub quota_type: String, + /// Optional durable reservation protocol. Omitted keeps the v1 default. + #[serde(default)] + pub reservation_protocol: Option, } #[derive(Debug, Deserialize)] @@ -56,6 +59,8 @@ struct CompatibleBucketQuotaRequest { size: Option, #[serde(default, alias = "quotatype")] quota_type: Option, + #[serde(default)] + reservation_protocol: Option, } #[derive(Debug, Serialize)] @@ -101,6 +106,7 @@ fn parse_set_bucket_quota_request(body: &[u8]) -> Result Result 0) .or_else(|| request.quota.filter(|quota| *quota > 0)), quota_type: request.quota_type.unwrap_or_else(default_quota_type), + reservation_protocol: request.reservation_protocol, }) } @@ -307,7 +314,16 @@ impl Operation for SetBucketQuotaHandler { None }; - let quota = BucketQuota::new(request.quota); + let quota = match request.reservation_protocol { + None | Some(crate::admin::storage_api::bucket::quota::QUOTA_RESERVATION_PROTOCOL_V1) => { + BucketQuota::new(request.quota) + } + Some(crate::admin::storage_api::bucket::quota::QUOTA_RESERVATION_PROTOCOL_V2) => match request.quota { + Some(limit) => BucketQuota::new_sharded(limit), + None => BucketQuota::new(None), + }, + Some(_) => return Err(s3_error!(InvalidArgument, "unsupported quota reservation protocol")), + }; let metadata_sys_lock = bucket_metadata_from_context() .ok_or_else(|| s3_error!(InternalError, "{}", rustfs_config::QUOTA_METADATA_SYSTEM_ERROR_MSG))?; @@ -741,6 +757,7 @@ mod tests { assert_eq!(request.quota, Some(1073741824)); assert_eq!(request.quota_type, "hard"); + assert_eq!(request.reservation_protocol, None); } #[test] @@ -752,6 +769,15 @@ mod tests { assert_eq!(request.quota_type, "hard"); } + #[test] + fn parse_set_bucket_quota_request_accepts_sharded_protocol() { + let request = parse_set_bucket_quota_request(br#"{"quota":1073741824,"reservation_protocol":2,"quota_type":"hard"}"#) + .expect("parse sharded quota request"); + + assert_eq!(request.quota, Some(1073741824)); + assert_eq!(request.reservation_protocol, Some(2)); + } + #[test] fn compat_bucket_quota_response_uses_external_field_names() { let quota = BucketQuota::new(Some(1024)); diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 3f56bcb2d..08ea72948 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -537,6 +537,7 @@ pub(crate) mod quota { 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) use super::ecstore_bucket::quota::{QUOTA_RESERVATION_PROTOCOL_V1, QUOTA_RESERVATION_PROTOCOL_V2}; } pub(crate) mod replication { diff --git a/rustfs/tests/connect_inventory.rs b/rustfs/tests/connect_inventory.rs index 292f5f5eb..a2044cd51 100644 --- a/rustfs/tests/connect_inventory.rs +++ b/rustfs/tests/connect_inventory.rs @@ -47,6 +47,7 @@ const ORGANIZATION_UID: &str = "0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70"; const CLUSTER_UID: &str = "0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81"; const DEVICE_UID: &str = "0198f4b0-3c00-7e30-8f41-4a5b6c7d8e92"; const SNAPSHOT_UID: &str = "0198f4b0-4d00-7f40-9051-5b6c7d8e9fa3"; +const INVENTORY_STATUS_WAIT_TIMEOUT: Duration = Duration::from_secs(10); fn safe_tempdir() -> tempfile::TempDir { tempfile::tempdir_in(env!("CARGO_MANIFEST_DIR")).expect("safe temporary directory") @@ -302,7 +303,7 @@ async fn wait_for( status: &mut watch::Receiver, predicate: impl Fn(&InventoryStatus) -> bool, ) -> InventoryStatus { - tokio::time::timeout(Duration::from_secs(3), async { + tokio::time::timeout(INVENTORY_STATUS_WAIT_TIMEOUT, async { loop { let current = status.borrow_and_update().clone(); if predicate(¤t) { diff --git a/scripts/error-other-format-baseline.txt b/scripts/error-other-format-baseline.txt index 7d2909a59..967a87143 100644 --- a/scripts/error-other-format-baseline.txt +++ b/scripts/error-other-format-baseline.txt @@ -14,7 +14,6 @@ 22|crates/ecstore/src/bucket/metadata_sys.rs 3|crates/ecstore/src/bucket/msgp_decode.rs 1|crates/ecstore/src/bucket/object_lock/objectlock_sys.rs -3|crates/ecstore/src/bucket/quota/reservation.rs 1|crates/ecstore/src/bucket/replication/replication_object_config.rs 2|crates/ecstore/src/bucket/replication/replication_pool.rs 2|crates/ecstore/src/bucket/replication/replication_target_boundary.rs