feat(quota): add sharded durable reservation protocol (#8402)

* feat(quota): add sharded durable reservation protocol

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(quota): preserve typed task error context

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(quota): initialize all durable reservation shards

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(quota): persist bootstrap shard eligibility

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* perf(quota): reuse allocator bootstrap state

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(connect): widen inventory startup wait

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(quota): reserve protocol v3 for unknown fixtures

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(quota): reload allocator after concurrent credit refills

---------

Co-authored-by: zhi22915 <qiuzgang@gmail.com>
Co-authored-by: overtrue <anzhengchao@gmail.com>

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;

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<u64> {
         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(&quota).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());

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 {

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.

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());
     }

rustfs/src/admin/handlers/quota.rs

@@ -46,6 +46,9 @@ pub struct SetBucketQuotaRequest {
     pub quota: Option<u64>,
     #[serde(default = "default_quota_type")]
     pub quota_type: String,
+    /// Optional durable reservation protocol. Omitted keeps the v1 default.
+    #[serde(default)]
+    pub reservation_protocol: Option<u32>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -56,6 +59,8 @@ struct CompatibleBucketQuotaRequest {
     size: Option<u64>,
     #[serde(default, alias = "quotatype")]
     quota_type: Option<String>,
+    #[serde(default)]
+    reservation_protocol: Option<u32>,
 }
 
 #[derive(Debug, Serialize)]
@@ -101,6 +106,7 @@ fn parse_set_bucket_quota_request(body: &[u8]) -> Result<SetBucketQuotaRequest,
         return Ok(SetBucketQuotaRequest {
             quota: None,
             quota_type: default_quota_type(),
+            reservation_protocol: None,
         });
     }
 
@@ -113,6 +119,7 @@ fn parse_set_bucket_quota_request(body: &[u8]) -> Result<SetBucketQuotaRequest,
             .filter(|quota| *quota > 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));

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 {

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<InventoryStatus>,
     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(&current) {

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