fix(ecstore): retain namespace read locks for streaming GETs to prevent quota-bypass (#5699)

* fix(ecstore): retain locks for streaming GETs

* fix(rpc): keep disabled snapshot leases lint-clean

rustfs/src/storage/rpc/node_service.rs

@@ -380,7 +380,6 @@ mod metrics;
 pub struct NodeService {
     local_peer: LocalPeerS3Client,
     context: Option<Arc<runtime_sources::AppContext>>,
-    snapshot_lease_expiry: disk::SnapshotLeaseExpiryScheduler,
 }
 
 impl std::fmt::Debug for NodeService {
@@ -399,11 +398,7 @@ pub fn make_server() -> NodeService {
 
 pub fn make_server_for_context(context: Option<Arc<runtime_sources::AppContext>>) -> NodeService {
     let local_peer = LocalPeerS3Client::new(None, None);
-    NodeService {
-        local_peer,
-        context,
-        snapshot_lease_expiry: disk::SnapshotLeaseExpiryScheduler::new(),
-    }
+    NodeService { local_peer, context }
 }
 
 #[derive(Clone, Debug, Default)]
@@ -2170,7 +2165,7 @@ mod tests {
         previous_scanner_activity_response, remove_heal_control_replay, scanner_activity_response, stop_rebalance_response,
     };
     use crate::storage::rpc::node_service::heal::heal_topology_fingerprint;
-    use crate::storage::storage_api::rpc_consumer::node_service::{HealBucketInfo, HealEndpoint};
+    use crate::storage::storage_api::rpc_consumer::node_service::{DiskError, HealBucketInfo, HealEndpoint};
     use crate::storage::storage_api::set_tonic_canonical_body_digest;
     use crate::storage::storage_api::{
         Endpoint,
@@ -2910,6 +2905,52 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn snapshot_lease_acquire_and_renew_handlers_fail_closed() {
+        let service = make_server();
+        let disk = "http://node-a:9000/data/rustfs0".to_string();
+
+        let mut acquire = Request::new(SnapshotLeaseRequest {
+            disk: disk.clone(),
+            volume: "v".into(),
+            path: "p".into(),
+            ttl_ms: 60_000,
+        });
+        let acquire_body =
+            rustfs_protos::canonical_snapshot_lease_request_body(acquire.get_ref()).expect("acquire request body should encode");
+        set_tonic_canonical_body_digest(&mut acquire, &acquire_body).expect("acquire digest metadata should encode");
+        mark_v2_authenticated(&mut acquire);
+        let acquire = service
+            .acquire_snapshot_lease(acquire)
+            .await
+            .expect("disabled acquire should return a protocol response")
+            .into_inner();
+
+        let mut renew = Request::new(SnapshotLeaseRenewRequest {
+            disk,
+            volume: "v".into(),
+            path: "p".into(),
+            token: vec![1; 16].into(),
+            ttl_ms: 60_000,
+        });
+        let renew_body = rustfs_protos::canonical_snapshot_lease_renew_request_body(renew.get_ref())
+            .expect("renew request body should encode");
+        set_tonic_canonical_body_digest(&mut renew, &renew_body).expect("renew digest metadata should encode");
+        mark_v2_authenticated(&mut renew);
+        let renew = service
+            .renew_snapshot_lease(renew)
+            .await
+            .expect("disabled renew should return a protocol response")
+            .into_inner();
+
+        for response in [acquire, renew] {
+            assert!(!response.success);
+            assert!(response.token.is_empty());
+            assert_eq!(response.protocol_version, 1);
+            assert_eq!(response.error, Some(DiskError::UnsupportedDisk.into()));
+        }
+    }
+
     #[tokio::test]
     async fn heal_control_requires_body_bound_auth_before_topology_validation() {
         let service = make_heal_control_server();

rustfs/src/storage/rpc/node_service/disk.rs

@@ -14,9 +14,8 @@
 
 use super::NodeService;
 use crate::storage::storage_api::rpc_consumer::node_service::{
-    BatchReadVersionReq, BatchReadVersionResp, DeleteOptions, DiskError, DiskInfoOptions, DiskStore, FileInfoVersions,
-    ReadMultipleReq, ReadMultipleResp, ReadOptions, StorageDiskRpcExt as _, UpdateMetadataOpts,
-    validate_batch_read_version_item_count,
+    BatchReadVersionReq, BatchReadVersionResp, DeleteOptions, DiskError, DiskInfoOptions, FileInfoVersions, ReadMultipleReq,
+    ReadMultipleResp, ReadOptions, StorageDiskRpcExt as _, UpdateMetadataOpts, validate_batch_read_version_item_count,
 };
 use crate::storage::storage_api::runtime_sources_consumer::runtime_sources;
 use crate::storage::storage_api::{PartTransactionAction, SnapshotLeaseToken, verify_tonic_mutation_body_digest};
@@ -29,9 +28,7 @@ use rustfs_io_metrics::internode_metrics::{
 };
 use rustfs_protos::proto_gen::node_service::*;
 use serde::de::DeserializeOwned;
-use std::{collections::HashMap, io::Cursor, time::Duration};
-use tokio::sync::mpsc;
-use tokio_util::time::DelayQueue;
+use std::io::Cursor;
 use tonic::{Request, Response, Status};
 use tracing::debug;
 
@@ -39,87 +36,15 @@ use tracing::debug;
 const MSGPACK_ENCODE_CAPACITY_HINT: usize = 512;
 const FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT: usize = 1024;
 const SNAPSHOT_LEASE_PROTOCOL_VERSION: u32 = 1;
-const SNAPSHOT_LEASE_MIN_TTL: Duration = Duration::from_secs(5);
-const SNAPSHOT_LEASE_MAX_TTL: Duration = Duration::from_secs(5 * 60);
 
-struct SnapshotLeaseExpiry {
-    disk: DiskStore,
-    volume: String,
-    path: String,
-    token: SnapshotLeaseToken,
-}
-
-pub(super) struct SnapshotLeaseExpiryScheduler {
-    tx: mpsc::UnboundedSender<SnapshotLeaseExpiryCommand>,
-}
-
-enum SnapshotLeaseExpiryCommand {
-    Schedule(SnapshotLeaseExpiry, Duration),
-    Cancel(SnapshotLeaseToken),
-}
-
-impl SnapshotLeaseExpiryScheduler {
-    pub(super) fn new() -> Self {
-        let (tx, mut rx) = mpsc::unbounded_channel();
-        tokio::spawn(async move {
-            let mut expirations: DelayQueue<SnapshotLeaseExpiry> = DelayQueue::new();
-            let mut keys = HashMap::new();
-            loop {
-                tokio::select! {
-                    Some(command) = rx.recv() => {
-                        match command {
-                            SnapshotLeaseExpiryCommand::Schedule(expiry, ttl) => {
-                                if let Some(key) = keys.remove(&expiry.token) {
-                                    expirations.remove(&key);
-                                }
-                                let token = expiry.token;
-                                let key = expirations.insert(expiry, ttl);
-                                keys.insert(token, key);
-                            }
-                            SnapshotLeaseExpiryCommand::Cancel(token) => {
-                                if let Some(key) = keys.remove(&token) {
-                                    expirations.remove(&key);
-                                }
-                            }
-                        }
-                    }
-                    Some(expired) = futures_util::StreamExt::next(&mut expirations), if !expirations.is_empty() => {
-                        let expiry = expired.into_inner();
-                        keys.remove(&expiry.token);
-                        let _ = expiry
-                            .disk
-                            .release_snapshot_lease(&expiry.volume, &expiry.path, expiry.token)
-                            .await;
-                    }
-                    else => break,
-                }
-            }
-        });
-        Self { tx }
-    }
-
-    fn schedule(&self, expiry: SnapshotLeaseExpiry, ttl: Duration) -> Result<(), SnapshotLeaseExpiry> {
-        self.tx
-            .send(SnapshotLeaseExpiryCommand::Schedule(expiry, ttl))
-            .map_err(|err| match err.0 {
-                SnapshotLeaseExpiryCommand::Schedule(expiry, _) => expiry,
-                SnapshotLeaseExpiryCommand::Cancel(_) => unreachable!(),
-            })
-    }
-
-    fn cancel(&self, token: SnapshotLeaseToken) {
-        let _ = self.tx.send(SnapshotLeaseExpiryCommand::Cancel(token));
+fn snapshot_lease_disabled_response() -> SnapshotLeaseResponse {
+    SnapshotLeaseResponse {
+        success: false,
+        token: Bytes::new(),
+        protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
+        error: Some(DiskError::UnsupportedDisk.into()),
     }
 }
-
-fn snapshot_lease_ttl(ttl_ms: u64) -> Result<Duration, Status> {
-    let ttl = Duration::from_millis(ttl_ms);
-    if !(SNAPSHOT_LEASE_MIN_TTL..=SNAPSHOT_LEASE_MAX_TTL).contains(&ttl) {
-        return Err(Status::invalid_argument("snapshot lease TTL is outside the supported range"));
-    }
-    Ok(ttl)
-}
-
 fn decode_msgpack_or_json<T: DeserializeOwned>(
     binary: &[u8],
     json: &str,
@@ -274,47 +199,7 @@ impl NodeService {
             rustfs_protos::canonical_snapshot_lease_request_body(request.get_ref()),
             "acquire_snapshot_lease",
         )?;
-        let request = request.into_inner();
-        let ttl = snapshot_lease_ttl(request.ttl_ms)?;
-        let Some(disk) = self.find_disk(&request.disk).await else {
-            return Ok(Response::new(SnapshotLeaseResponse {
-                success: false,
-                token: Bytes::new(),
-                protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
-                error: Some(DiskError::other("cannot find disk").into()),
-            }));
-        };
-        match disk.acquire_snapshot_lease(&request.volume, &request.path).await {
-            Ok(token) => {
-                if let Err(expiry) = self.snapshot_lease_expiry.schedule(
-                    SnapshotLeaseExpiry {
-                        disk,
-                        volume: request.volume,
-                        path: request.path,
-                        token,
-                    },
-                    ttl,
-                ) {
-                    let _ = expiry
-                        .disk
-                        .release_snapshot_lease(&expiry.volume, &expiry.path, expiry.token)
-                        .await;
-                    return Err(Status::internal("snapshot lease expiry scheduler is unavailable"));
-                }
-                Ok(Response::new(SnapshotLeaseResponse {
-                    success: true,
-                    token: Bytes::copy_from_slice(token.as_bytes()),
-                    protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
-                    error: None,
-                }))
-            }
-            Err(err) => Ok(Response::new(SnapshotLeaseResponse {
-                success: false,
-                token: Bytes::new(),
-                protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
-                error: Some(err.into()),
-            })),
-        }
+        Ok(Response::new(snapshot_lease_disabled_response()))
     }
 
     pub(super) async fn handle_renew_snapshot_lease(
@@ -326,50 +211,7 @@ impl NodeService {
             rustfs_protos::canonical_snapshot_lease_renew_request_body(request.get_ref()),
             "renew_snapshot_lease",
         )?;
-        let request = request.into_inner();
-        let ttl = snapshot_lease_ttl(request.ttl_ms)?;
-        let token =
-            SnapshotLeaseToken::from_slice(&request.token).map_err(|_| Status::invalid_argument("invalid lease token"))?;
-        let Some(disk) = self.find_disk(&request.disk).await else {
-            return Ok(Response::new(SnapshotLeaseResponse {
-                success: false,
-                token: Bytes::new(),
-                protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
-                error: Some(DiskError::other("cannot find disk").into()),
-            }));
-        };
-        match disk.renew_snapshot_lease(&request.volume, &request.path, token).await {
-            Ok(renewed) => {
-                if let Err(expiry) = self.snapshot_lease_expiry.schedule(
-                    SnapshotLeaseExpiry {
-                        disk,
-                        volume: request.volume,
-                        path: request.path,
-                        token: renewed,
-                    },
-                    ttl,
-                ) {
-                    let _ = expiry
-                        .disk
-                        .release_snapshot_lease(&expiry.volume, &expiry.path, expiry.token)
-                        .await;
-                    return Err(Status::internal("snapshot lease expiry scheduler is unavailable"));
-                }
-                self.snapshot_lease_expiry.cancel(token);
-                Ok(Response::new(SnapshotLeaseResponse {
-                    success: true,
-                    token: Bytes::copy_from_slice(renewed.as_bytes()),
-                    protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
-                    error: None,
-                }))
-            }
-            Err(err) => Ok(Response::new(SnapshotLeaseResponse {
-                success: false,
-                token: Bytes::new(),
-                protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
-                error: Some(err.into()),
-            })),
-        }
+        Ok(Response::new(snapshot_lease_disabled_response()))
     }
 
     pub(super) async fn handle_release_snapshot_lease(
@@ -391,13 +233,10 @@ impl NodeService {
             }));
         };
         match disk.release_snapshot_lease(&request.volume, &request.path, token).await {
-            Ok(()) => {
-                self.snapshot_lease_expiry.cancel(token);
-                Ok(Response::new(SnapshotLeaseMutationResponse {
-                    success: true,
-                    error: None,
-                }))
-            }
+            Ok(()) => Ok(Response::new(SnapshotLeaseMutationResponse {
+                success: true,
+                error: None,
+            })),
             Err(err) => Ok(Response::new(SnapshotLeaseMutationResponse {
                 success: false,
                 error: Some(err.into()),
@@ -1608,10 +1447,10 @@ impl NodeService {
 #[cfg(test)]
 mod tests {
     use super::{
-        SNAPSHOT_LEASE_MAX_TTL, SNAPSHOT_LEASE_MIN_TTL, compat_response_json, decode_msgpack_or_json,
-        encode_batch_read_version_response_payloads, encode_msgpack, encode_msgpack_named,
-        encode_read_multiple_response_payloads, snapshot_lease_ttl,
+        compat_response_json, decode_msgpack_or_json, encode_batch_read_version_response_payloads, encode_msgpack,
+        encode_msgpack_named, encode_read_multiple_response_payloads, snapshot_lease_disabled_response,
     };
+    use crate::storage::storage_api::DiskError;
     use crate::storage::storage_api::ReadMultipleResp;
     use crate::storage::storage_api::rpc_consumer::node_service::BatchReadVersionResp;
     use rustfs_io_metrics::internode_metrics::global_internode_metrics;
@@ -1624,11 +1463,14 @@ mod tests {
     }
 
     #[test]
-    fn snapshot_lease_ttl_rejects_values_outside_server_bounds() {
-        assert!(snapshot_lease_ttl(4_999).is_err());
-        assert_eq!(snapshot_lease_ttl(5_000).unwrap(), SNAPSHOT_LEASE_MIN_TTL);
-        assert_eq!(snapshot_lease_ttl(300_000).unwrap(), SNAPSHOT_LEASE_MAX_TTL);
-        assert!(snapshot_lease_ttl(300_001).is_err());
+    fn snapshot_lease_acquire_and_renew_fail_closed() {
+        let response = snapshot_lease_disabled_response();
+        let expected_error = DiskError::UnsupportedDisk.into();
+
+        assert!(!response.success);
+        assert!(response.token.is_empty());
+        assert_eq!(response.protocol_version, 1);
+        assert_eq!(response.error, Some(expected_error));
     }
 
     #[test]

rustfs/src/storage/storage_api.rs

@@ -1099,8 +1099,6 @@ pub(crate) trait StorageDiskRpcExt {
     ) -> DiskResult<()>;
     async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult<bytes::Bytes>;
     async fn delete_paths(&self, volume: &str, paths: &[String]) -> DiskResult<()>;
-    async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> DiskResult<SnapshotLeaseToken>;
-    async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> DiskResult<SnapshotLeaseToken>;
     async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> DiskResult<()>;
     async fn stat_volume(&self, volume: &str) -> DiskResult<VolumeInfo>;
     async fn list_volumes(&self) -> DiskResult<Vec<VolumeInfo>>;
@@ -1228,14 +1226,6 @@ where
         ecstore_disk::DiskAPI::delete_paths(self, volume, paths).await
     }
 
-    async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> DiskResult<SnapshotLeaseToken> {
-        ecstore_disk::DiskAPI::acquire_snapshot_lease(self, volume, path).await
-    }
-
-    async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> DiskResult<SnapshotLeaseToken> {
-        ecstore_disk::DiskAPI::renew_snapshot_lease(self, volume, path, token).await
-    }
-
     async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> DiskResult<()> {
         ecstore_disk::DiskAPI::release_snapshot_lease(self, volume, path, token).await
     }