fix(ecstore): retain GET lock for missing lease disk

Cargo.lock

@@ -11829,6 +11829,7 @@ dependencies = [
  "futures-util",
  "libc",
  "pin-project-lite",
+ "slab",
  "tokio",
 ]
 

crates/e2e_test/src/reliant/grpc_lock_server.rs

@@ -23,7 +23,8 @@ use rustfs_protos::{
     proto_gen::node_service::{
         BatchGenerallyLockRequest, BatchGenerallyLockResponse, BatchReadVersionRequest, BatchReadVersionResponse,
         GenerallyLockRequest, GenerallyLockResponse, GenerallyLockResult, PingRequest, PingResponse,
-        node_service_server::NodeService,
+        SnapshotLeaseMutationResponse, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest,
+        SnapshotLeaseResponse, node_service_server::NodeService,
     },
 };
 use std::pin::Pin;
@@ -104,6 +105,27 @@ impl NodeService for MinimalLockNodeService {
         Err(Status::unimplemented("MinimalLockNodeService only supports lock RPCs"))
     }
 
+    async fn acquire_snapshot_lease(
+        &self,
+        _request: Request<SnapshotLeaseRequest>,
+    ) -> Result<Response<SnapshotLeaseResponse>, Status> {
+        Err(Status::unimplemented("MinimalLockNodeService only supports lock RPCs"))
+    }
+
+    async fn renew_snapshot_lease(
+        &self,
+        _request: Request<SnapshotLeaseRenewRequest>,
+    ) -> Result<Response<SnapshotLeaseResponse>, Status> {
+        Err(Status::unimplemented("MinimalLockNodeService only supports lock RPCs"))
+    }
+
+    async fn release_snapshot_lease(
+        &self,
+        _request: Request<SnapshotLeaseReleaseRequest>,
+    ) -> Result<Response<SnapshotLeaseMutationResponse>, Status> {
+        Err(Status::unimplemented("MinimalLockNodeService only supports lock RPCs"))
+    }
+
     async fn lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
         let request = request.into_inner();
         let args: LockRequest = match serde_json::from_str(&request.args) {

crates/ecstore/src/api/mod.rs

@@ -305,7 +305,8 @@ pub mod disk {
         CheckPartsResp, DeleteOptions, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, DiskStore,
         FileInfoVersions, FileReader, FileWriter, HEALING_MARKER_PATH, NsScannerOpenRequest, OldCurrentSize,
         PartTransactionAction, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp,
-        STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk, validate_batch_read_version_item_count,
+        STORAGE_FORMAT_FILE, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk,
+        validate_batch_read_version_item_count,
     };
     pub use bytes::Bytes;
     pub use endpoint::Endpoint;

crates/ecstore/src/cluster/rpc/remote_disk.rs

@@ -25,7 +25,7 @@ use crate::disk::error::{Error, Result};
 use crate::disk::{
     BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation,
     DiskOption, FileInfoVersions, FileReader, FileWriter, PartTransactionAction, ReadMultipleReq, ReadMultipleResp, ReadOptions,
-    RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, batch_read_version_one_by_one,
+    RenameDataResp, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, batch_read_version_one_by_one,
     disk_store::{
         DEFAULT_RUSTFS_DRIVE_ACTIVE_MONITORING, ENV_RUSTFS_DRIVE_ACTIVE_MONITORING, SKIP_IF_SUCCESS_BEFORE,
         get_drive_active_check_interval, get_drive_active_check_timeout, get_drive_disk_info_timeout, get_drive_list_dir_timeout,
@@ -50,8 +50,9 @@ use rustfs_protos::proto_gen::node_service::{
     DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, ListVolumesRequest,
     MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest, ReadMetadataRequest,
     ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest,
-    RenameFileRequest, SettlePartTransactionRequest, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest,
-    WriteAllRequest, WriteMetadataRequest, node_service_client::NodeServiceClient,
+    RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
+    SnapshotLeaseRequest, SnapshotLeaseResponse, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest,
+    WriteMetadataRequest, node_service_client::NodeServiceClient,
 };
 use serde::{Serialize, de::DeserializeOwned};
 use std::{
@@ -100,6 +101,18 @@ const LOG_COMPONENT_ECSTORE: &str = "ecstore";
 const LOG_SUBSYSTEM_REMOTE_DISK: &str = "remote_disk";
 const EVENT_REMOTE_DISK_HEALTH: &str = "remote_disk_health";
 const EVENT_REMOTE_DISK_RPC: &str = "remote_disk_rpc";
+const SNAPSHOT_LEASE_PROTOCOL_VERSION: u32 = 1;
+pub const REMOTE_SNAPSHOT_LEASE_TTL: Duration = Duration::from_secs(60);
+
+fn snapshot_lease_token_from_response(response: SnapshotLeaseResponse) -> Result<SnapshotLeaseToken> {
+    if !response.success {
+        return Err(response.error.unwrap_or_default().into());
+    }
+    if response.protocol_version != SNAPSHOT_LEASE_PROTOCOL_VERSION {
+        return Err(Error::other("remote snapshot lease protocol is incompatible"));
+    }
+    SnapshotLeaseToken::from_slice(&response.token)
+}
 
 /// Bind a mutating disk RPC to its canonical body: the digest lands in the request metadata, and
 /// the signing interceptor folds it (plus a replay-protected nonce) into the v2 signature scope
@@ -1784,6 +1797,81 @@ impl DiskAPI for RemoteDisk {
         .await
     }
 
+    async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> Result<SnapshotLeaseToken> {
+        self.execute_with_timeout(
+            || async {
+                let mut client = self
+                    .get_client()
+                    .await
+                    .map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
+                let mut request = Request::new(SnapshotLeaseRequest {
+                    disk: self.endpoint.to_string(),
+                    volume: volume.to_string(),
+                    path: path.to_string(),
+                    ttl_ms: u64::try_from(REMOTE_SNAPSHOT_LEASE_TTL.as_millis())
+                        .map_err(|_| Error::other("snapshot lease TTL cannot be represented"))?,
+                });
+                let canonical_body = rustfs_protos::canonical_snapshot_lease_request_body(request.get_ref());
+                attach_mutation_body_digest(&mut request, canonical_body, "acquire_snapshot_lease")?;
+                let response = client.acquire_snapshot_lease(request).await?.into_inner();
+                snapshot_lease_token_from_response(response)
+            },
+            get_max_timeout_duration(),
+        )
+        .await
+    }
+
+    async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<SnapshotLeaseToken> {
+        self.execute_with_timeout(
+            || async {
+                let mut client = self
+                    .get_client()
+                    .await
+                    .map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
+                let mut request = Request::new(SnapshotLeaseRenewRequest {
+                    disk: self.endpoint.to_string(),
+                    volume: volume.to_string(),
+                    path: path.to_string(),
+                    token: token.as_bytes().to_vec().into(),
+                    ttl_ms: u64::try_from(REMOTE_SNAPSHOT_LEASE_TTL.as_millis())
+                        .map_err(|_| Error::other("snapshot lease TTL cannot be represented"))?,
+                });
+                let canonical_body = rustfs_protos::canonical_snapshot_lease_renew_request_body(request.get_ref());
+                attach_mutation_body_digest(&mut request, canonical_body, "renew_snapshot_lease")?;
+                let response = client.renew_snapshot_lease(request).await?.into_inner();
+                snapshot_lease_token_from_response(response)
+            },
+            get_max_timeout_duration(),
+        )
+        .await
+    }
+
+    async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<()> {
+        self.execute_with_timeout(
+            || async {
+                let mut client = self
+                    .get_client()
+                    .await
+                    .map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
+                let mut request = Request::new(SnapshotLeaseReleaseRequest {
+                    disk: self.endpoint.to_string(),
+                    volume: volume.to_string(),
+                    path: path.to_string(),
+                    token: token.as_bytes().to_vec().into(),
+                });
+                let canonical_body = rustfs_protos::canonical_snapshot_lease_release_request_body(request.get_ref());
+                attach_mutation_body_digest(&mut request, canonical_body, "release_snapshot_lease")?;
+                let response = client.release_snapshot_lease(request).await?.into_inner();
+                if !response.success {
+                    return Err(response.error.unwrap_or_default().into());
+                }
+                Ok(())
+            },
+            get_max_timeout_duration(),
+        )
+        .await
+    }
+
     #[tracing::instrument(level = "trace", skip_all)]
     async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
         trace!(
@@ -2930,6 +3018,34 @@ mod tests {
 
     static INIT: Once = Once::new();
 
+    #[test]
+    fn snapshot_lease_response_requires_current_protocol_and_valid_token() {
+        let token = SnapshotLeaseToken::new();
+        let response = SnapshotLeaseResponse {
+            success: true,
+            token: token.as_bytes().to_vec().into(),
+            protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
+            error: None,
+        };
+        assert_eq!(snapshot_lease_token_from_response(response).unwrap(), token);
+
+        let incompatible = SnapshotLeaseResponse {
+            success: true,
+            token: token.as_bytes().to_vec().into(),
+            protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION + 1,
+            error: None,
+        };
+        assert!(snapshot_lease_token_from_response(incompatible).is_err());
+
+        let malformed = SnapshotLeaseResponse {
+            success: true,
+            token: Bytes::from_static(b"not-a-uuid"),
+            protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
+            error: None,
+        };
+        assert!(snapshot_lease_token_from_response(malformed).is_err());
+    }
+
     #[test]
     fn list_volumes_decode_rejects_a_malformed_entry() {
         let valid = serde_json::to_string(&VolumeInfo {

crates/ecstore/src/disk/disk_store.rs

@@ -1365,6 +1365,14 @@ impl DiskAPI for LocalDiskWrapper {
         .await
     }
 
+    async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<SnapshotLeaseToken> {
+        self.track_disk_health(
+            || async { self.disk.renew_snapshot_lease(volume, path, token).await },
+            get_max_timeout_duration(),
+        )
+        .await
+    }
+
     async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result<DataDirDeleteStatus> {
         self.track_disk_health(
             || async { self.disk.delete_data_dir(volume, path, opts).await },

crates/ecstore/src/disk/local.rs

@@ -7908,6 +7908,23 @@ impl DiskAPI for LocalDisk {
         }
     }
 
+    async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<SnapshotLeaseToken> {
+        let key = SnapshotLeaseKey {
+            volume: volume.to_string(),
+            path: path.to_string(),
+        };
+        let mut registry = self.snapshot_leases.lock().await;
+        let Some(entry) = registry.entries.get_mut(&key) else {
+            return Err(DiskError::FileNotFound);
+        };
+        if entry.deleting || !entry.tokens.remove(&token) {
+            return Err(DiskError::FileNotFound);
+        }
+        let renewed = SnapshotLeaseToken::new();
+        entry.tokens.insert(renewed);
+        Ok(renewed)
+    }
+
     async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result<DataDirDeleteStatus> {
         let key = SnapshotLeaseKey {
             volume: volume.to_string(),
@@ -14957,6 +14974,13 @@ mod test {
             .acquire_snapshot_lease(volume, &data_dir)
             .await
             .expect("second lease should be acquired");
+        let renewed = disk
+            .renew_snapshot_lease(volume, &data_dir, first)
+            .await
+            .expect("first lease should renew atomically");
+        disk.release_snapshot_lease(volume, &data_dir, first)
+            .await
+            .expect("the superseded token should be idempotent");
         let status = disk
             .delete_data_dir(
                 volume,
@@ -14976,9 +15000,9 @@ mod test {
             Bytes::from_static(b"later")
         );
 
-        disk.release_snapshot_lease(volume, &data_dir, first)
+        disk.release_snapshot_lease(volume, &data_dir, renewed)
             .await
-            .expect("first lease release should succeed");
+            .expect("renewed lease release should succeed");
         assert!(
             disk.read_all(volume, &first_part).await.is_ok(),
             "one remaining lease must keep the data directory"

crates/ecstore/src/disk/mod.rs

@@ -79,6 +79,18 @@ impl SnapshotLeaseToken {
     pub fn new() -> Self {
         Self(Uuid::new_v4())
     }
+
+    pub fn from_slice(bytes: &[u8]) -> Result<Self> {
+        let uuid = Uuid::from_slice(bytes).map_err(|_| Error::other("invalid snapshot lease token"))?;
+        if uuid.is_nil() {
+            return Err(Error::other("invalid snapshot lease token"));
+        }
+        Ok(Self(uuid))
+    }
+
+    pub fn as_bytes(&self) -> &[u8; 16] {
+        self.0.as_bytes()
+    }
 }
 
 impl Default for SnapshotLeaseToken {
@@ -284,6 +296,13 @@ impl DiskAPI for Disk {
         }
     }
 
+    async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<SnapshotLeaseToken> {
+        match self {
+            Disk::Local(local_disk) => local_disk.renew_snapshot_lease(volume, path, token).await,
+            Disk::Remote(remote_disk) => remote_disk.renew_snapshot_lease(volume, path, token).await,
+        }
+    }
+
     async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result<DataDirDeleteStatus> {
         match self {
             Disk::Local(local_disk) => local_disk.delete_data_dir(volume, path, opts).await,
@@ -694,6 +713,9 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     async fn release_snapshot_lease(&self, _volume: &str, _path: &str, _token: SnapshotLeaseToken) -> Result<()> {
         Err(Error::other("snapshot leases are not supported by this disk"))
     }
+    async fn renew_snapshot_lease(&self, _volume: &str, _path: &str, _token: SnapshotLeaseToken) -> Result<SnapshotLeaseToken> {
+        Err(Error::other("snapshot leases are not supported by this disk"))
+    }
     async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result<DataDirDeleteStatus> {
         self.delete(volume, path, opts).await?;
         Ok(DataDirDeleteStatus::Deleted)

crates/ecstore/src/set_disk/mod.rs

@@ -104,7 +104,7 @@ use crate::{
     disk::{
         CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskOption, DiskStore, FileInfoVersions,
         RUSTFS_META_BUCKET, RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions,
-        UpdateMetadataOpts, endpoint::Endpoint, error::DiskError, format::FormatV3, new_disk,
+        SnapshotLeaseToken, UpdateMetadataOpts, endpoint::Endpoint, error::DiskError, format::FormatV3, new_disk,
     },
     error::{StorageError, to_object_err},
     object_api::{GetObjectReader, ObjectInfo, PutObjReader},
@@ -116,6 +116,7 @@ use bytes::Bytes;
 use bytesize::ByteSize;
 use chrono::Utc;
 use futures::future::join_all;
+use futures::task::AtomicWaker;
 use glob::Pattern;
 use http::HeaderMap;
 use md5::{Digest as Md5Digest, Md5};
@@ -164,7 +165,7 @@ use std::hash::{BuildHasher, Hash, Hasher};
 use std::mem::{self};
 use std::pin::Pin;
 use std::sync::OnceLock;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::task::{Context, Poll};
 use std::time::{Instant, SystemTime, UNIX_EPOCH};
 use std::{
@@ -359,6 +360,163 @@ struct SetDiskLockGuardedReader {
     guard: Option<ObjectLockDiagGuard>,
 }
 
+#[derive(Clone)]
+struct SnapshotLease {
+    disk: DiskStore,
+    volume: String,
+    path: String,
+    token: SnapshotLeaseToken,
+}
+
+struct SnapshotLeaseState {
+    leases: parking_lot::Mutex<Vec<SnapshotLease>>,
+    renewal_failed: AtomicBool,
+    renewal_waker: AtomicWaker,
+    cancel: CancellationToken,
+    runtime: tokio::runtime::Handle,
+}
+
+impl Drop for SnapshotLeaseState {
+    fn drop(&mut self) {
+        self.cancel.cancel();
+        let leases = mem::take(&mut *self.leases.lock());
+        if leases.is_empty() {
+            return;
+        }
+        self.runtime.spawn(async move {
+            join_all(leases.into_iter().map(|lease| async move {
+                lease
+                    .disk
+                    .release_snapshot_lease(&lease.volume, &lease.path, lease.token)
+                    .await
+            }))
+            .await;
+        });
+    }
+}
+
+#[derive(Clone)]
+struct SnapshotLeaseHandle(Arc<SnapshotLeaseState>);
+
+impl SnapshotLeaseHandle {
+    fn new(leases: Vec<SnapshotLease>) -> Self {
+        let state = Arc::new(SnapshotLeaseState {
+            leases: parking_lot::Mutex::new(leases),
+            renewal_failed: AtomicBool::new(false),
+            renewal_waker: AtomicWaker::new(),
+            cancel: CancellationToken::new(),
+            runtime: tokio::runtime::Handle::current(),
+        });
+        let weak = Arc::downgrade(&state);
+        let cancel = state.cancel.clone();
+        tokio::spawn(async move {
+            let renew_interval = crate::cluster::rpc::remote_disk::REMOTE_SNAPSHOT_LEASE_TTL / 3;
+            loop {
+                tokio::select! {
+                    _ = cancel.cancelled() => return,
+                    _ = tokio::time::sleep(renew_interval) => {}
+                }
+                let Some(state) = weak.upgrade() else {
+                    return;
+                };
+                let leases = state.leases.lock().clone();
+                let results = join_all(
+                    leases
+                        .iter()
+                        .map(|lease| lease.disk.renew_snapshot_lease(&lease.volume, &lease.path, lease.token)),
+                )
+                .await;
+                let mut current = state.leases.lock();
+                let mut failed = false;
+                for (lease, result) in current.iter_mut().zip(results) {
+                    match result {
+                        Ok(token) => lease.token = token,
+                        Err(_) => failed = true,
+                    }
+                }
+                drop(current);
+                if failed {
+                    state.renewal_failed.store(true, Ordering::Release);
+                    state.renewal_waker.wake();
+                    return;
+                }
+            }
+        });
+        Self(state)
+    }
+
+    fn renewal_failed(&self) -> bool {
+        self.0.renewal_failed.load(Ordering::Acquire)
+    }
+}
+
+struct SnapshotLeaseReader {
+    inner: Box<dyn AsyncRead + Unpin + Send + Sync>,
+    lease: Option<SnapshotLeaseHandle>,
+}
+
+impl AsyncRead for SnapshotLeaseReader {
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+        let Some(lease) = self.lease.as_ref() else {
+            return Pin::new(&mut self.inner).poll_read(cx, buf);
+        };
+        if lease.renewal_failed() {
+            self.lease.take();
+            return Poll::Ready(Err(std::io::Error::other("snapshot lease renewal failed")));
+        }
+        lease.0.renewal_waker.register(cx.waker());
+        if lease.renewal_failed() {
+            self.lease.take();
+            return Poll::Ready(Err(std::io::Error::other("snapshot lease renewal failed")));
+        }
+        let filled_before = buf.filled().len();
+        let poll = Pin::new(&mut self.inner).poll_read(cx, buf);
+        if matches!(poll, Poll::Ready(Err(_)))
+            || matches!(poll, Poll::Ready(Ok(())) if buf.filled().len() == filled_before && buf.remaining() > 0)
+        {
+            self.lease.take();
+        }
+        poll
+    }
+}
+
+async fn acquire_snapshot_leases(
+    disks: &[Option<DiskStore>],
+    volume: &str,
+    path: &str,
+    read_quorum: usize,
+) -> Option<SnapshotLeaseHandle> {
+    let candidates = disks.iter().cloned().collect::<Option<Vec<_>>>()?;
+    if candidates.len() < read_quorum {
+        return None;
+    }
+    let results = join_all(candidates.iter().map(|disk| disk.acquire_snapshot_lease(volume, path))).await;
+    let mut leases = Vec::with_capacity(candidates.len());
+    let mut failed = false;
+    for (disk, result) in candidates.into_iter().zip(results) {
+        match result {
+            Ok(token) => leases.push(SnapshotLease {
+                disk,
+                volume: volume.to_string(),
+                path: path.to_string(),
+                token,
+            }),
+            Err(_) => failed = true,
+        }
+    }
+    if failed || leases.len() < read_quorum {
+        join_all(leases.into_iter().map(|lease| async move {
+            lease
+                .disk
+                .release_snapshot_lease(&lease.volume, &lease.path, lease.token)
+                .await
+        }))
+        .await;
+        return None;
+    }
+    Some(SnapshotLeaseHandle::new(leases))
+}
+
 impl AsyncRead for SetDiskLockGuardedReader {
     fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
         let had_capacity = buf.remaining() > 0;
@@ -374,15 +532,24 @@ impl AsyncRead for SetDiskLockGuardedReader {
 fn finish_set_disk_read_lock(
     mut reader: GetObjectReader,
     read_lock_guard: Option<ObjectLockDiagGuard>,
-    lock_optimization_enabled: bool,
+    snapshot_lease: Option<SnapshotLeaseHandle>,
     bucket: &str,
     object: &str,
 ) -> GetObjectReader {
-    if lock_optimization_enabled || reader.buffered_body.is_some() {
+    if reader.buffered_body.is_some() {
         release_materialized_read_lock(bucket, object, read_lock_guard);
         return reader;
     }
 
+    if let Some(lease) = snapshot_lease {
+        release_materialized_read_lock(bucket, object, read_lock_guard);
+        reader.stream = Box::new(SnapshotLeaseReader {
+            inner: reader.stream,
+            lease: Some(lease),
+        });
+        return reader;
+    }
+
     if let Some(guard) = read_lock_guard {
         reader.stream = Box::new(SetDiskLockGuardedReader {
             inner: reader.stream,
@@ -1024,8 +1191,9 @@ pub fn get_object_lock_diag_slow_hold_threshold() -> Duration {
 }
 
 /// Check if lock optimization is enabled.
-/// When enabled, fully materialized reads may release the read lock before
-/// returning to the caller. Streaming reads keep the lock until EOF or drop.
+/// Fully materialized reads release the read lock before returning. Streaming
+/// reads may replace it with data-directory snapshot leases when every
+/// candidate disk supports the lease protocol.
 ///
 /// **Note**: Cached via `OnceLock` in production — env var changes require
 /// process restart. In test builds the env var is read directly so that
@@ -5475,6 +5643,88 @@ mod tests {
         (dir, disk)
     }
 
+    #[tokio::test]
+    async fn snapshot_lease_acquisition_is_all_or_nothing() {
+        let (_dir1, disk1) = make_single_local_disk().await;
+        let (_dir2, disk2) = make_single_local_disk().await;
+        let bucket = "snapshot-lease-acquire";
+        let data_dir = "object/11111111-1111-1111-1111-111111111111";
+        let part = format!("{data_dir}/part.1");
+        disk1.make_volume(bucket).await.expect("first volume should be created");
+        disk2.make_volume(bucket).await.expect("second volume should be created");
+        disk1
+            .write_all(bucket, &part, Bytes::from_static(b"shard"))
+            .await
+            .expect("first shard should be written");
+
+        let lease = acquire_snapshot_leases(&[Some(disk1.clone()), Some(disk2)], bucket, data_dir, 1).await;
+        assert!(lease.is_none(), "one candidate missing snapshot data must retain the namespace lock");
+        assert_eq!(
+            disk1
+                .delete_data_dir(
+                    bucket,
+                    data_dir,
+                    DeleteOptions {
+                        recursive: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("released partial lease should not defer cleanup"),
+            crate::disk::DataDirDeleteStatus::Deleted
+        );
+    }
+
+    #[tokio::test]
+    async fn snapshot_lease_acquisition_rejects_unavailable_candidate() {
+        let (_dir, disk) = make_single_local_disk().await;
+        let bucket = "snapshot-lease-unavailable";
+        let data_dir = "object/11111111-1111-1111-1111-111111111111";
+        let part = format!("{data_dir}/part.1");
+        disk.make_volume(bucket).await.expect("volume should be created");
+        disk.write_all(bucket, &part, Bytes::from_static(b"shard"))
+            .await
+            .expect("shard should be written");
+
+        let lease = acquire_snapshot_leases(&[Some(disk.clone()), None], bucket, data_dir, 1).await;
+        assert!(lease.is_none(), "one unavailable candidate must retain the namespace lock");
+        assert_eq!(
+            disk.delete_data_dir(
+                bucket,
+                data_dir,
+                DeleteOptions {
+                    recursive: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("unavailable candidate fallback must not leave a lease"),
+            crate::disk::DataDirDeleteStatus::Deleted
+        );
+    }
+
+    #[tokio::test]
+    async fn snapshot_lease_reader_fails_when_renewal_fails() {
+        let state = Arc::new(SnapshotLeaseState {
+            leases: parking_lot::Mutex::new(Vec::new()),
+            renewal_failed: AtomicBool::new(true),
+            renewal_waker: AtomicWaker::new(),
+            cancel: CancellationToken::new(),
+            runtime: tokio::runtime::Handle::current(),
+        });
+        let mut reader = SnapshotLeaseReader {
+            inner: Box::new(Cursor::new(Bytes::from_static(b"payload"))),
+            lease: Some(SnapshotLeaseHandle(state)),
+        };
+        let mut body = Vec::new();
+        let err = reader
+            .read_to_end(&mut body)
+            .await
+            .expect_err("renewal failure must terminate the body");
+        assert_eq!(err.kind(), std::io::ErrorKind::Other);
+        assert!(body.is_empty());
+    }
+
     async fn make_set_disks_with(disks: Vec<Option<DiskStore>>) -> Arc<SetDisks> {
         let drive_count = disks.len();
         let endpoints = (0..drive_count)
@@ -9051,6 +9301,78 @@ mod tests {
         );
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn streaming_get_snapshot_survives_concurrent_overwrite() {
+        temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, Some("true"))], async {
+            let set_disks = make_local_bucket_test_set_disks().await;
+            let bucket = "snapshot-streaming-overwrite";
+            let object = "object";
+            let old_body = vec![0x41; 2 * 1024 * 1024];
+            let new_body = vec![0x42; old_body.len()];
+            let opts = ObjectOptions::default();
+
+            set_disks
+                .make_bucket(bucket, &MakeBucketOptions::default())
+                .await
+                .expect("bucket should be created");
+            let mut old_reader = PutObjReader::from_vec(old_body.clone());
+            set_disks
+                .put_object(bucket, object, &mut old_reader, &opts)
+                .await
+                .expect("old object should be written");
+
+            let mut snapshot = set_disks
+                .get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
+                .await
+                .expect("snapshot reader should open");
+            let overwrite_set = Arc::clone(&set_disks);
+            let overwrite_body = new_body.clone();
+            let overwrite_opts = opts.clone();
+            let overwrite = tokio::spawn(async move {
+                let mut reader = PutObjReader::from_vec(overwrite_body);
+                overwrite_set.put_object(bucket, object, &mut reader, &overwrite_opts).await
+            });
+            tokio::time::timeout(Duration::from_secs(5), overwrite)
+                .await
+                .expect("overwrite should not wait for the response body")
+                .expect("overwrite task should join")
+                .expect("overwrite should succeed");
+
+            let mut restored = Vec::new();
+            snapshot
+                .stream
+                .read_to_end(&mut restored)
+                .await
+                .expect("leased snapshot should remain readable");
+            assert_eq!(restored, old_body);
+
+            let mut latest = set_disks
+                .get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
+                .await
+                .expect("latest reader should open");
+            let mut latest_body = Vec::new();
+            latest
+                .stream
+                .read_to_end(&mut latest_body)
+                .await
+                .expect("latest object should remain readable");
+            assert_eq!(latest_body, new_body);
+
+            let cancelled = set_disks
+                .get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
+                .await
+                .expect("cancelled reader should open");
+            drop(cancelled);
+            let mut replacement = PutObjReader::from_vec(vec![0x43; 2 * 1024 * 1024]);
+            tokio::time::timeout(Duration::from_secs(5), set_disks.put_object(bucket, object, &mut replacement, &opts))
+                .await
+                .expect("reader drop must release its snapshot")
+                .expect("replacement after cancellation should succeed");
+        })
+        .await;
+    }
+
     #[tokio::test]
     async fn set_level_batched_large_put_get_restores_body() {
         const BATCHED_LARGE_SIZE: usize = 64 * 1024 * 1024;

crates/ecstore/src/set_disk/ops/object.rs

@@ -550,13 +550,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 &self.ctx.tier_config_mgr(),
             )
             .await?;
-            return Ok(finish_set_disk_read_lock(
-                gr,
-                read_lock_guard.take(),
-                lock_optimization_enabled,
-                bucket,
-                object,
-            ));
+            return Ok(finish_set_disk_read_lock(gr, read_lock_guard.take(), None, bucket, object));
         }
 
         // App-layer object data cache probe: metadata (etag/size) is resolved
@@ -683,6 +677,18 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
             return Ok(reader);
         }
 
+        let snapshot_lease = if lock_optimization_enabled {
+            match fi.data_dir.filter(|data_dir| !data_dir.is_nil()) {
+                Some(data_dir) => {
+                    let data_dir_path = format!("{object}/{data_dir}");
+                    acquire_snapshot_leases(&disks, bucket, &data_dir_path, fi.erasure.data_blocks).await
+                }
+                None => None,
+            }
+        } else {
+            None
+        };
+
         match codec_streaming_gate.decision {
             GetCodecStreamingDecision::Use => {
                 match Self::get_object_decode_reader_with_fileinfo(
@@ -711,13 +717,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                         // Carry the hook probe result so the app layer skips its
                         // now-redundant lookup on the streaming miss path (ODC-16).
                         reader.body_source = body_source;
-                        return Ok(finish_set_disk_read_lock(
-                            reader,
-                            read_lock_guard.take(),
-                            lock_optimization_enabled,
-                            bucket,
-                            object,
-                        ));
+                        return Ok(finish_set_disk_read_lock(reader, read_lock_guard.take(), snapshot_lease, bucket, object));
                     }
                     core::io_primitives::GetCodecStreamingReaderBuildOutcome::Fallback(reason) => {
                         record_get_codec_streaming_gate_decision(
@@ -756,15 +756,21 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         let set_index = self.set_index;
         let pool_index = self.pool_index;
         let skip_verify = opts.skip_verify_bitrot;
-        if lock_optimization_enabled {
+        let producer_snapshot_lease = snapshot_lease.clone();
+        if let Some(lease) = snapshot_lease {
             release_materialized_read_lock(&bucket, &object, read_lock_guard.take());
-            debug!(bucket, object, "Lock optimization: released read lock before streaming read");
+            reader.stream = Box::new(SnapshotLeaseReader {
+                inner: reader.stream,
+                lease: Some(lease),
+            });
+            debug!(bucket, object, "Lock optimization: replaced read lock with snapshot leases");
         }
 
-        // When lock optimization is disabled, keep the read-lock guard in the
-        // task so it lives for the duration of the streaming read.
+        // The producer shares the lease lifetime with the body so cancellation
+        // cannot release the snapshot while the duplex task is still unwinding.
         tokio::spawn(async move {
             let _guard = read_lock_guard;
+            let _snapshot_lease = producer_snapshot_lease;
             let mut writer = GetObjectDownstreamWriter::new(wd);
             // Do not wrap the entire read+write pipeline in `disk_read_timeout`.
             // `get_object_with_fileinfo` also waits on `writer`, so an outer timeout

crates/protos/src/generated/proto_gen/node_service.rs

@@ -484,6 +484,59 @@ pub struct DeletePathsResponse {
     pub error: ::core::option::Option<Error>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct SnapshotLeaseRequest {
+    #[prost(string, tag = "1")]
+    pub disk: ::prost::alloc::string::String,
+    #[prost(string, tag = "2")]
+    pub volume: ::prost::alloc::string::String,
+    #[prost(string, tag = "3")]
+    pub path: ::prost::alloc::string::String,
+    #[prost(uint64, tag = "4")]
+    pub ttl_ms: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct SnapshotLeaseRenewRequest {
+    #[prost(string, tag = "1")]
+    pub disk: ::prost::alloc::string::String,
+    #[prost(string, tag = "2")]
+    pub volume: ::prost::alloc::string::String,
+    #[prost(string, tag = "3")]
+    pub path: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "4")]
+    pub token: ::prost::bytes::Bytes,
+    #[prost(uint64, tag = "5")]
+    pub ttl_ms: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct SnapshotLeaseReleaseRequest {
+    #[prost(string, tag = "1")]
+    pub disk: ::prost::alloc::string::String,
+    #[prost(string, tag = "2")]
+    pub volume: ::prost::alloc::string::String,
+    #[prost(string, tag = "3")]
+    pub path: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "4")]
+    pub token: ::prost::bytes::Bytes,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct SnapshotLeaseResponse {
+    #[prost(bool, tag = "1")]
+    pub success: bool,
+    #[prost(bytes = "bytes", tag = "2")]
+    pub token: ::prost::bytes::Bytes,
+    #[prost(uint32, tag = "3")]
+    pub protocol_version: u32,
+    #[prost(message, optional, tag = "4")]
+    pub error: ::core::option::Option<Error>,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct SnapshotLeaseMutationResponse {
+    #[prost(bool, tag = "1")]
+    pub success: bool,
+    #[prost(message, optional, tag = "2")]
+    pub error: ::core::option::Option<Error>,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ReadMetadataRequest {
     #[prost(string, tag = "1")]
     pub disk: ::prost::alloc::string::String,
@@ -1595,6 +1648,51 @@ pub mod node_service_client {
                 .insert(GrpcMethod::new("node_service.NodeService", "Delete"));
             self.inner.unary(req, path, codec).await
         }
+        pub async fn acquire_snapshot_lease(
+            &mut self,
+            request: impl tonic::IntoRequest<super::SnapshotLeaseRequest>,
+        ) -> std::result::Result<tonic::Response<super::SnapshotLeaseResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/AcquireSnapshotLease");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.NodeService", "AcquireSnapshotLease"));
+            self.inner.unary(req, path, codec).await
+        }
+        pub async fn renew_snapshot_lease(
+            &mut self,
+            request: impl tonic::IntoRequest<super::SnapshotLeaseRenewRequest>,
+        ) -> std::result::Result<tonic::Response<super::SnapshotLeaseResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/RenewSnapshotLease");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.NodeService", "RenewSnapshotLease"));
+            self.inner.unary(req, path, codec).await
+        }
+        pub async fn release_snapshot_lease(
+            &mut self,
+            request: impl tonic::IntoRequest<super::SnapshotLeaseReleaseRequest>,
+        ) -> std::result::Result<tonic::Response<super::SnapshotLeaseMutationResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/ReleaseSnapshotLease");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.NodeService", "ReleaseSnapshotLease"));
+            self.inner.unary(req, path, codec).await
+        }
         pub async fn verify_file(
             &mut self,
             request: impl tonic::IntoRequest<super::VerifyFileRequest>,
@@ -2784,6 +2882,18 @@ pub mod node_service_server {
             &self,
             request: tonic::Request<super::DeleteRequest>,
         ) -> std::result::Result<tonic::Response<super::DeleteResponse>, tonic::Status>;
+        async fn acquire_snapshot_lease(
+            &self,
+            request: tonic::Request<super::SnapshotLeaseRequest>,
+        ) -> std::result::Result<tonic::Response<super::SnapshotLeaseResponse>, tonic::Status>;
+        async fn renew_snapshot_lease(
+            &self,
+            request: tonic::Request<super::SnapshotLeaseRenewRequest>,
+        ) -> std::result::Result<tonic::Response<super::SnapshotLeaseResponse>, tonic::Status>;
+        async fn release_snapshot_lease(
+            &self,
+            request: tonic::Request<super::SnapshotLeaseReleaseRequest>,
+        ) -> std::result::Result<tonic::Response<super::SnapshotLeaseMutationResponse>, tonic::Status>;
         async fn verify_file(
             &self,
             request: tonic::Request<super::VerifyFileRequest>,
@@ -3426,6 +3536,90 @@ pub mod node_service_server {
                     };
                     Box::pin(fut)
                 }
+                "/node_service.NodeService/AcquireSnapshotLease" => {
+                    #[allow(non_camel_case_types)]
+                    struct AcquireSnapshotLeaseSvc<T: NodeService>(pub Arc<T>);
+                    impl<T: NodeService> tonic::server::UnaryService<super::SnapshotLeaseRequest> for AcquireSnapshotLeaseSvc<T> {
+                        type Response = super::SnapshotLeaseResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::SnapshotLeaseRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move { <T as NodeService>::acquire_snapshot_lease(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = AcquireSnapshotLeaseSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                "/node_service.NodeService/RenewSnapshotLease" => {
+                    #[allow(non_camel_case_types)]
+                    struct RenewSnapshotLeaseSvc<T: NodeService>(pub Arc<T>);
+                    impl<T: NodeService> tonic::server::UnaryService<super::SnapshotLeaseRenewRequest> for RenewSnapshotLeaseSvc<T> {
+                        type Response = super::SnapshotLeaseResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::SnapshotLeaseRenewRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move { <T as NodeService>::renew_snapshot_lease(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = RenewSnapshotLeaseSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                "/node_service.NodeService/ReleaseSnapshotLease" => {
+                    #[allow(non_camel_case_types)]
+                    struct ReleaseSnapshotLeaseSvc<T: NodeService>(pub Arc<T>);
+                    impl<T: NodeService> tonic::server::UnaryService<super::SnapshotLeaseReleaseRequest> for ReleaseSnapshotLeaseSvc<T> {
+                        type Response = super::SnapshotLeaseMutationResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::SnapshotLeaseReleaseRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move { <T as NodeService>::release_snapshot_lease(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = ReleaseSnapshotLeaseSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
                 "/node_service.NodeService/VerifyFile" => {
                     #[allow(non_camel_case_types)]
                     struct VerifyFileSvc<T: NodeService>(pub Arc<T>);

crates/protos/src/lib.rs

@@ -451,6 +451,10 @@ impl CanonicalBodyBuilder {
         self.body.push(u8::from(field));
     }
 
+    fn push_u64(&mut self, field: u64) {
+        self.body.extend_from_slice(&field.to_be_bytes());
+    }
+
     fn push_count(&mut self, count: usize) -> Result<(), std::num::TryFromIntError> {
         self.body.extend_from_slice(&u64::try_from(count)?.to_be_bytes());
         Ok(())
@@ -575,6 +579,40 @@ pub fn canonical_delete_paths_request_body(
     Ok(body.finish())
 }
 
+pub fn canonical_snapshot_lease_request_body(
+    request: &proto_gen::node_service::SnapshotLeaseRequest,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-snapshot-lease-request-v1\0");
+    body.push_str(&request.disk)?;
+    body.push_str(&request.volume)?;
+    body.push_str(&request.path)?;
+    body.push_u64(request.ttl_ms);
+    Ok(body.finish())
+}
+
+pub fn canonical_snapshot_lease_renew_request_body(
+    request: &proto_gen::node_service::SnapshotLeaseRenewRequest,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-snapshot-lease-renew-request-v1\0");
+    body.push_str(&request.disk)?;
+    body.push_str(&request.volume)?;
+    body.push_str(&request.path)?;
+    body.push_bytes(&request.token)?;
+    body.push_u64(request.ttl_ms);
+    Ok(body.finish())
+}
+
+pub fn canonical_snapshot_lease_release_request_body(
+    request: &proto_gen::node_service::SnapshotLeaseReleaseRequest,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-snapshot-lease-release-request-v1\0");
+    body.push_str(&request.disk)?;
+    body.push_str(&request.volume)?;
+    body.push_str(&request.path)?;
+    body.push_bytes(&request.token)?;
+    Ok(body.finish())
+}
+
 pub fn canonical_rename_file_request_body(
     request: &proto_gen::node_service::RenameFileRequest,
 ) -> Result<Vec<u8>, std::num::TryFromIntError> {
@@ -662,7 +700,8 @@ mod disk_mutation_canonical_tests {
     use super::proto_gen::node_service::{
         DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, MakeVolumeRequest,
         MakeVolumesRequest, PreparePartTransactionRequest, RenameDataRequest, RenameFileRequest, RenamePartRequest,
-        SettlePartTransactionRequest, UpdateMetadataRequest, WriteAllRequest, WriteMetadataRequest,
+        SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest,
+        UpdateMetadataRequest, WriteAllRequest, WriteMetadataRequest,
     };
     use super::*;
 
@@ -1020,6 +1059,58 @@ mod disk_mutation_canonical_tests {
         );
     }
 
+    #[test]
+    fn snapshot_lease_canonical_bodies_bind_every_field() {
+        let acquire = SnapshotLeaseRequest {
+            disk: "d".into(),
+            volume: "v".into(),
+            path: "p".into(),
+            ttl_ms: 60_000,
+        };
+        let mut acquire_bodies = vec![canonical_snapshot_lease_request_body(&acquire).unwrap()];
+        for mutate in [
+            |r: &mut SnapshotLeaseRequest| r.disk = "d2".into(),
+            |r: &mut SnapshotLeaseRequest| r.volume = "v2".into(),
+            |r: &mut SnapshotLeaseRequest| r.path = "p2".into(),
+            |r: &mut SnapshotLeaseRequest| r.ttl_ms = 60_001,
+        ] {
+            let mut request = acquire.clone();
+            mutate(&mut request);
+            acquire_bodies.push(canonical_snapshot_lease_request_body(&request).unwrap());
+        }
+        assert_all_distinct(&acquire_bodies);
+
+        let renew = SnapshotLeaseRenewRequest {
+            disk: "d".into(),
+            volume: "v".into(),
+            path: "p".into(),
+            token: vec![1; 16].into(),
+            ttl_ms: 60_000,
+        };
+        let mut changed_token = renew.clone();
+        changed_token.token = vec![2; 16].into();
+        let mut changed_ttl = renew.clone();
+        changed_ttl.ttl_ms += 1;
+        assert_all_distinct(&[
+            canonical_snapshot_lease_renew_request_body(&renew).unwrap(),
+            canonical_snapshot_lease_renew_request_body(&changed_token).unwrap(),
+            canonical_snapshot_lease_renew_request_body(&changed_ttl).unwrap(),
+        ]);
+
+        let release = SnapshotLeaseReleaseRequest {
+            disk: "d".into(),
+            volume: "v".into(),
+            path: "p".into(),
+            token: vec![1; 16].into(),
+        };
+        let mut changed_release = release.clone();
+        changed_release.token = vec![2; 16].into();
+        assert_ne!(
+            canonical_snapshot_lease_release_request_body(&release).unwrap(),
+            canonical_snapshot_lease_release_request_body(&changed_release).unwrap()
+        );
+    }
+
     #[test]
     fn disk_mutation_canonical_domains_are_distinct_per_message() {
         // The same field values must never authenticate one RPC's request as another's.

crates/protos/src/node.proto

@@ -342,6 +342,40 @@ message DeletePathsResponse {
   optional Error error = 2;
 }
 
+message SnapshotLeaseRequest {
+  string disk = 1;
+  string volume = 2;
+  string path = 3;
+  uint64 ttl_ms = 4;
+}
+
+message SnapshotLeaseRenewRequest {
+  string disk = 1;
+  string volume = 2;
+  string path = 3;
+  bytes token = 4;
+  uint64 ttl_ms = 5;
+}
+
+message SnapshotLeaseReleaseRequest {
+  string disk = 1;
+  string volume = 2;
+  string path = 3;
+  bytes token = 4;
+}
+
+message SnapshotLeaseResponse {
+  bool success = 1;
+  bytes token = 2;
+  uint32 protocol_version = 3;
+  optional Error error = 4;
+}
+
+message SnapshotLeaseMutationResponse {
+  bool success = 1;
+  optional Error error = 2;
+}
+
 message ReadMetadataRequest {
   string disk = 1;
   string volume = 2;
@@ -966,6 +1000,9 @@ service NodeService {
   rpc ReadAll(ReadAllRequest) returns (ReadAllResponse) {};
   rpc WriteAll(WriteAllRequest) returns (WriteAllResponse) {};
   rpc Delete(DeleteRequest) returns (DeleteResponse) {};
+  rpc AcquireSnapshotLease(SnapshotLeaseRequest) returns (SnapshotLeaseResponse) {};
+  rpc RenewSnapshotLease(SnapshotLeaseRenewRequest) returns (SnapshotLeaseResponse) {};
+  rpc ReleaseSnapshotLease(SnapshotLeaseReleaseRequest) returns (SnapshotLeaseMutationResponse) {};
   rpc VerifyFile(VerifyFileRequest) returns (VerifyFileResponse) {};
   rpc ReadParts(ReadPartsRequest) returns (ReadPartsResponse) {};
   rpc CheckParts(CheckPartsRequest) returns (CheckPartsResponse) {};

rustfs/Cargo.toml

@@ -124,7 +124,7 @@ tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "sig
 tokio-rustls = { workspace = true, default-features = false, features = ["logging", "tls12", "aws-lc-rs"] }
 aws-sdk-s3 = { workspace = true, default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] }
 tokio-stream.workspace = true
-tokio-util = { workspace = true, features = ["io", "compat"] }
+tokio-util = { workspace = true, features = ["io", "compat", "time"] }
 tonic = { workspace = true, features = ["gzip", "deflate"] }
 tower = { workspace = true, features = ["timeout"] }
 tower-http = { workspace = true, features = ["trace", "compression-full", "cors", "catch-panic", "timeout", "limit", "request-id", "add-extension"] }

rustfs/src/storage/rpc/node_service.rs

@@ -343,6 +343,7 @@ 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 {
@@ -361,7 +362,11 @@ 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 }
+    NodeService {
+        local_peer,
+        context,
+        snapshot_lease_expiry: disk::SnapshotLeaseExpiryScheduler::new(),
+    }
 }
 
 #[derive(Clone, Debug, Default)]
@@ -1124,6 +1129,24 @@ impl Node for NodeService {
     async fn delete_paths(&self, request: Request<DeletePathsRequest>) -> Result<Response<DeletePathsResponse>, Status> {
         self.handle_delete_paths(request).await
     }
+    async fn acquire_snapshot_lease(
+        &self,
+        request: Request<SnapshotLeaseRequest>,
+    ) -> Result<Response<SnapshotLeaseResponse>, Status> {
+        self.handle_acquire_snapshot_lease(request).await
+    }
+    async fn renew_snapshot_lease(
+        &self,
+        request: Request<SnapshotLeaseRenewRequest>,
+    ) -> Result<Response<SnapshotLeaseResponse>, Status> {
+        self.handle_renew_snapshot_lease(request).await
+    }
+    async fn release_snapshot_lease(
+        &self,
+        request: Request<SnapshotLeaseReleaseRequest>,
+    ) -> Result<Response<SnapshotLeaseMutationResponse>, Status> {
+        self.handle_release_snapshot_lease(request).await
+    }
     async fn read_metadata(&self, request: Request<ReadMetadataRequest>) -> Result<Response<ReadMetadataResponse>, Status> {
         self.handle_read_metadata(request).await
     }

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

@@ -14,11 +14,12 @@
 
 use super::NodeService;
 use crate::storage::storage_api::rpc_consumer::node_service::{
-    BatchReadVersionReq, BatchReadVersionResp, DeleteOptions, DiskError, DiskInfoOptions, FileInfoVersions, ReadMultipleReq,
-    ReadMultipleResp, ReadOptions, StorageDiskRpcExt as _, UpdateMetadataOpts, validate_batch_read_version_item_count,
+    BatchReadVersionReq, BatchReadVersionResp, DeleteOptions, DiskError, DiskInfoOptions, DiskStore, 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, verify_tonic_mutation_body_digest};
+use crate::storage::storage_api::{PartTransactionAction, SnapshotLeaseToken, verify_tonic_mutation_body_digest};
 use bytes::Bytes;
 use rustfs_filemeta::FileInfo;
 use rustfs_io_metrics::internode_metrics::{
@@ -28,13 +29,96 @@ use rustfs_io_metrics::internode_metrics::{
 };
 use rustfs_protos::proto_gen::node_service::*;
 use serde::de::DeserializeOwned;
-use std::io::Cursor;
+use std::{collections::HashMap, io::Cursor, time::Duration};
+use tokio::sync::mpsc;
+use tokio_util::time::DelayQueue;
 use tonic::{Request, Response, Status};
 use tracing::debug;
 
 /// Initial capacity hint (bytes) for msgpack encode buffers, sized to cover a typical single-
 /// version `FileInfo` without repeated growth reallocations. Larger payloads still grow as needed.
 const MSGPACK_ENCODE_CAPACITY_HINT: usize = 512;
+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_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],
@@ -165,6 +249,146 @@ fn encode_batch_read_version_response_payloads(
 }
 
 impl NodeService {
+    pub(super) async fn handle_acquire_snapshot_lease(
+        &self,
+        request: Request<SnapshotLeaseRequest>,
+    ) -> Result<Response<SnapshotLeaseResponse>, Status> {
+        verify_disk_mutation_digest(
+            &request,
+            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()),
+            })),
+        }
+    }
+
+    pub(super) async fn handle_renew_snapshot_lease(
+        &self,
+        request: Request<SnapshotLeaseRenewRequest>,
+    ) -> Result<Response<SnapshotLeaseResponse>, Status> {
+        verify_disk_mutation_digest(
+            &request,
+            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()),
+            })),
+        }
+    }
+
+    pub(super) async fn handle_release_snapshot_lease(
+        &self,
+        request: Request<SnapshotLeaseReleaseRequest>,
+    ) -> Result<Response<SnapshotLeaseMutationResponse>, Status> {
+        verify_disk_mutation_digest(
+            &request,
+            rustfs_protos::canonical_snapshot_lease_release_request_body(request.get_ref()),
+            "release_snapshot_lease",
+        )?;
+        let request = request.into_inner();
+        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(SnapshotLeaseMutationResponse {
+                success: false,
+                error: Some(DiskError::other("cannot find disk").into()),
+            }));
+        };
+        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,
+                }))
+            }
+            Err(err) => Ok(Response::new(SnapshotLeaseMutationResponse {
+                success: false,
+                error: Some(err.into()),
+            })),
+        }
+    }
+
     pub(super) async fn handle_disk_info(&self, request: Request<DiskInfoRequest>) -> Result<Response<DiskInfoResponse>, Status> {
         let request = request.into_inner();
         if let Some(disk) = self.find_disk(&request.disk).await {
@@ -1368,8 +1592,9 @@ impl NodeService {
 #[cfg(test)]
 mod tests {
     use super::{
-        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_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,
     };
     use crate::storage::storage_api::ReadMultipleResp;
     use crate::storage::storage_api::rpc_consumer::node_service::BatchReadVersionResp;
@@ -1382,6 +1607,14 @@ mod tests {
         count: u32,
     }
 
+    #[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());
+    }
+
     #[test]
     fn decode_msgpack_or_json_prefers_binary_payload() {
         let payload = SamplePayload {

rustfs/src/storage/storage_api.rs

@@ -431,7 +431,7 @@ pub(crate) mod ecstore_disk {
     pub(crate) use rustfs_ecstore::api::disk::{
         BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskStore,
         FileInfoVersions, FileReader, FileWriter, OldCurrentSize, PartTransactionAction, RUSTFS_META_BUCKET, ReadMultipleReq,
-        ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
+        ReadMultipleResp, ReadOptions, RenameDataResp, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
         get_object_disk_read_timeout, validate_batch_read_version_item_count,
     };
     pub(crate) use rustfs_ecstore::api::disk::{endpoint, error, error_reduce};
@@ -606,6 +606,7 @@ pub(crate) type ExpiryState = ecstore_bucket::lifecycle::bucket_lifecycle_ops::E
 pub(crate) type FileInfoVersions = ecstore_disk::FileInfoVersions;
 pub(crate) type FileReader = ecstore_disk::FileReader;
 pub(crate) type FileWriter = ecstore_disk::FileWriter;
+pub(crate) type SnapshotLeaseToken = ecstore_disk::SnapshotLeaseToken;
 pub(crate) type FS = super::ecfs::FS;
 pub(crate) type HashReader = ecstore_rio::HashReader;
 pub(crate) type InstanceContext = ecstore_runtime::InstanceContext;
@@ -1075,6 +1076,9 @@ 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>>;
     async fn make_volume(&self, volume: &str) -> DiskResult<()>;
@@ -1201,6 +1205,18 @@ 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
+    }
+
     async fn stat_volume(&self, volume: &str) -> DiskResult<VolumeInfo> {
         ecstore_disk::DiskAPI::stat_volume(self, volume).await
     }