fix(ecstore): release GET locks at content length

.config/nextest.toml

@@ -55,6 +55,12 @@ test-group = 'ecstore-serial-flaky'
 filter = 'package(rustfs-ecstore) & test(/^set_disk::ops::multipart::tests::crash_consistency::/)'
 test-group = 'ecstore-serial-flaky'
 
+# This regression test builds three multipart objects on a four-disk set and
+# coordinates live GET/DELETE races under both lock modes.
+[[profile.default.overrides]]
+filter = 'package(rustfs-ecstore) & test(multipart_get_keeps_delete_blocked_at_part_boundary_in_both_lock_modes)'
+test-group = 'ecstore-serial-flaky'
+
 # Serialize the durable manual-transition checkpoint test across nextest's
 # process boundary; it mutates bucket lifecycle metadata and is not quarantined.
 [[profile.default.overrides]]
@@ -136,6 +142,10 @@ test-group = 'e2e-reliability'
 filter = 'package(rustfs-ecstore) & test(/^set_disk::ops::multipart::tests::crash_consistency::/)'
 test-group = 'ecstore-serial-flaky'
 
+[[profile.ci.overrides]]
+filter = 'package(rustfs-ecstore) & test(multipart_get_keeps_delete_blocked_at_part_boundary_in_both_lock_modes)'
+test-group = 'ecstore-serial-flaky'
+
 # Serialize the durable manual-transition checkpoint test under the ci profile
 # too. No retries: failures stay visible.
 [[profile.ci.overrides]]

crates/ecstore/src/set_disk/mod.rs

@@ -357,6 +357,7 @@ impl Drop for ObjectLockDiagGuard {
 struct SetDiskLockGuardedReader {
     inner: Box<dyn AsyncRead + Unpin + Send + Sync>,
     guard: Option<ObjectLockDiagGuard>,
+    remaining: Option<usize>,
 }
 
 impl AsyncRead for SetDiskLockGuardedReader {
@@ -364,8 +365,23 @@ impl AsyncRead for SetDiskLockGuardedReader {
         let had_capacity = buf.remaining() > 0;
         let filled_before = buf.filled().len();
         let poll = Pin::new(&mut self.inner).poll_read(cx, buf);
-        if had_capacity && matches!(poll, Poll::Ready(Ok(()))) && buf.filled().len() == filled_before {
-            self.guard.take();
+        match &poll {
+            Poll::Ready(Ok(())) if buf.filled().len() > filled_before => {
+                let produced = buf.filled().len() - filled_before;
+                if let Some(remaining) = self.remaining.as_mut() {
+                    *remaining = remaining.saturating_sub(produced);
+                    if *remaining == 0 {
+                        self.guard.take();
+                    }
+                }
+            }
+            Poll::Ready(Ok(())) if had_capacity => {
+                self.guard.take();
+            }
+            Poll::Ready(Err(_)) => {
+                self.guard.take();
+            }
+            _ => {}
         }
         poll
     }
@@ -374,19 +390,20 @@ impl AsyncRead for SetDiskLockGuardedReader {
 fn finish_set_disk_read_lock(
     mut reader: GetObjectReader,
     read_lock_guard: Option<ObjectLockDiagGuard>,
-    lock_optimization_enabled: bool,
     bucket: &str,
     object: &str,
 ) -> GetObjectReader {
-    if lock_optimization_enabled || reader.buffered_body.is_some() {
+    if reader.buffered_body.is_some() || reader.object_info.size == 0 {
         release_materialized_read_lock(bucket, object, read_lock_guard);
         return reader;
     }
 
     if let Some(guard) = read_lock_guard {
+        let remaining = usize::try_from(reader.object_info.size).ok();
         reader.stream = Box::new(SetDiskLockGuardedReader {
             inner: reader.stream,
             guard: Some(guard),
+            remaining,
         });
     }
     reader

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

@@ -2147,6 +2147,163 @@ mod tests {
         }
     }
 
+    async fn put_two_part_object(set_disks: &Arc<SetDisks>, bucket: &str, object: &str, first: &[u8], second: &[u8]) {
+        use crate::storage_api_contracts::multipart::MultipartOperations as _;
+
+        let upload = set_disks
+            .new_multipart_upload(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("multipart upload should be created");
+        let mut completed = Vec::new();
+        for (part_number, payload) in [(1, first), (2, second)] {
+            let payload_len = i64::try_from(payload.len()).expect("test payload length should fit i64");
+            let mut reader = PutObjReader::new(
+                HashReader::from_stream(Cursor::new(payload.to_vec()), payload_len, payload_len, None, None, false)
+                    .expect("part hash reader should be created"),
+            );
+            let part = set_disks
+                .put_object_part(bucket, object, &upload.upload_id, part_number, &mut reader, &ObjectOptions::default())
+                .await
+                .expect("multipart part should be written");
+            completed.push(CompletePart {
+                part_num: part.part_num,
+                etag: part.etag,
+                ..Default::default()
+            });
+        }
+        set_disks
+            .clone()
+            .complete_multipart_upload(bucket, object, &upload.upload_id, completed, &ObjectOptions::default())
+            .await
+            .expect("multipart upload should complete");
+    }
+
+    async fn assert_get_blocks_delete_at_part_boundary(
+        set_disks: &Arc<SetDisks>,
+        bucket: &str,
+        object: &'static str,
+        range: Option<HTTPRangeSpec>,
+        expected: Vec<u8>,
+    ) {
+        use crate::set_disk::read::get_part_boundary_barrier;
+        use http::HeaderMap;
+        use tokio::io::AsyncReadExt as _;
+
+        let barrier = get_part_boundary_barrier::arm(bucket, object, 0);
+        let mut reader = set_disks
+            .get_object_reader(bucket, object, range, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("GET should return a body reader");
+        let body_complete = Arc::new(Notify::new());
+        let body_complete_task = Arc::clone(&body_complete);
+        let allow_reader_drop = Arc::new(Notify::new());
+        let allow_reader_drop_task = Arc::clone(&allow_reader_drop);
+        let expected_len = expected.len();
+        let read = tokio::spawn(async move {
+            let mut body = vec![0_u8; expected_len];
+            reader.stream.read_exact(&mut body).await?;
+            body_complete_task.notify_one();
+            allow_reader_drop_task.notified().await;
+            Ok::<_, std::io::Error>(body)
+        });
+        barrier.wait_until_paused().await;
+
+        let delete_set = Arc::clone(set_disks);
+        let delete_bucket = bucket.to_string();
+        let delete_started = Arc::new(Notify::new());
+        let delete_started_task = Arc::clone(&delete_started);
+        let mut delete = tokio::spawn(async move {
+            delete_started_task.notify_one();
+            delete_set
+                .delete_object(&delete_bucket, object, ObjectOptions::default())
+                .await
+        });
+        delete_started.notified().await;
+        tokio::task::yield_now().await;
+        assert!(!delete.is_finished(), "DELETE must wait behind the streaming GET read lock");
+
+        barrier.release();
+        body_complete.notified().await;
+        let delete_result = tokio::time::timeout(Duration::from_secs(2), &mut delete).await;
+        allow_reader_drop.notify_one();
+        delete_result
+            .expect("DELETE must acquire the lock after the exact Content-Length without waiting for reader drop")
+            .expect("DELETE task should not panic")
+            .expect("DELETE should proceed after GET delivers its exact Content-Length");
+        let body = read
+            .await
+            .expect("GET task should not panic")
+            .expect("GET stream should deliver its exact Content-Length");
+        assert_eq!(body.len(), expected.len(), "successful GET body must match the requested length");
+        assert_eq!(body, expected, "DELETE must not truncate or mix the multipart snapshot");
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn multipart_get_keeps_delete_blocked_at_part_boundary_in_both_lock_modes() {
+        use crate::set_disk::read::get_part_boundary_barrier;
+        use tokio::io::AsyncReadExt as _;
+
+        for enabled in ["false", "true"] {
+            temp_env::async_with_vars(
+                [
+                    (rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, Some(enabled)),
+                    (ENV_RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE, Some("false")),
+                ],
+                async {
+                    let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+                    let bucket = format!("get-delete-part-boundary-{enabled}");
+                    let object = "object";
+                    make_bucket_on_all(&disk_stores, &bucket).await;
+
+                    let first = vec![0x41; GLOBAL_MIN_PART_SIZE.as_u64() as usize];
+                    let second = vec![0x42; 64 * 1024];
+                    let expected: Vec<u8> = first.iter().chain(&second).copied().collect();
+                    put_two_part_object(&set_disks, &bucket, object, &first, &second).await;
+                    assert_get_blocks_delete_at_part_boundary(&set_disks, &bucket, object, None, expected).await;
+
+                    let range_object = "range-object";
+                    put_two_part_object(&set_disks, &bucket, range_object, &first, &second).await;
+                    let range_start = i64::try_from(first.len() - 4096).expect("range start should fit i64");
+                    let range_end = i64::try_from(first.len() + 4095).expect("range end should fit i64");
+                    let range = HTTPRangeSpec {
+                        is_suffix_length: false,
+                        start: range_start,
+                        end: range_end,
+                    };
+                    let range_expected = first[first.len() - 4096..].iter().chain(&second[..4096]).copied().collect();
+                    assert_get_blocks_delete_at_part_boundary(&set_disks, &bucket, range_object, Some(range), range_expected)
+                        .await;
+
+                    let cancelled_object = "cancelled-object";
+                    put_two_part_object(&set_disks, &bucket, cancelled_object, &first, &second).await;
+                    let cancel_barrier = get_part_boundary_barrier::arm(&bucket, cancelled_object, 0);
+                    let mut cancelled_reader = set_disks
+                        .get_object_reader(&bucket, cancelled_object, None, HeaderMap::new(), &ObjectOptions::default())
+                        .await
+                        .expect("cancelled GET should return a body reader");
+                    let cancelled_read = tokio::spawn(async move {
+                        let mut body = Vec::new();
+                        cancelled_reader.stream.read_to_end(&mut body).await
+                    });
+                    cancel_barrier.wait_until_paused().await;
+                    cancelled_read.abort();
+                    cancelled_read.await.expect_err("aborted GET task should report cancellation");
+                    cancel_barrier.release();
+
+                    tokio::time::timeout(
+                        Duration::from_secs(10),
+                        set_disks.delete_object(&bucket, cancelled_object, ObjectOptions::default()),
+                    )
+                    .await
+                    .expect("DELETE should promptly acquire the lock after GET cancellation")
+                    .expect("DELETE after cancellation should succeed");
+                },
+            )
+            .await;
+        }
+    }
+
     async fn stage_upload_with_create_opts(
         set_disks: &Arc<SetDisks>,
         bucket: &str,

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

@@ -426,13 +426,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(), bucket, object));
         }
 
         // App-layer object data cache probe: metadata (etag/size) is resolved
@@ -587,13 +581,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(), bucket, object));
                     }
                     core::io_primitives::GetCodecStreamingReaderBuildOutcome::Fallback(reason) => {
                         record_get_codec_streaming_gate_decision(
@@ -632,13 +620,8 @@ 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 {
-            release_materialized_read_lock(&bucket, &object, read_lock_guard.take());
-            debug!(bucket, object, "Lock optimization: released read lock before streaming read");
-        }
-
-        // When lock optimization is disabled, keep the read-lock guard in the
-        // task so it lives for the duration of the streaming read.
+        // Keep the read lock until the producer reaches EOF or observes that
+        // the downstream reader was cancelled.
         tokio::spawn(async move {
             let _guard = read_lock_guard;
             let mut writer = GetObjectDownstreamWriter::new(wd);

crates/ecstore/src/set_disk/read.rs

@@ -1117,6 +1117,10 @@ impl SetDisks {
 
             total_read += part_length;
             part_offset = 0;
+            #[cfg(test)]
+            if current_part < last_part_index {
+                get_part_boundary_barrier::checkpoint(bucket, object, current_part).await;
+            }
         }
 
         // debug!("read end");
@@ -1820,6 +1824,79 @@ fn is_get_object_metadata_cache_request_eligible(bucket: &str, opts: &ObjectOpti
     get_object_metadata_cache_request_bypass_reason(bucket, opts, read_data).is_none()
 }
 
+#[cfg(test)]
+pub(in crate::set_disk) mod get_part_boundary_barrier {
+    use std::collections::HashMap;
+    use std::sync::{Arc, Mutex, OnceLock};
+    use tokio::sync::Notify;
+
+    struct Armed {
+        part_index: usize,
+        arrived: Arc<Notify>,
+        release: Arc<Notify>,
+    }
+
+    fn registry() -> &'static Mutex<HashMap<(String, String), Armed>> {
+        static REGISTRY: OnceLock<Mutex<HashMap<(String, String), Armed>>> = OnceLock::new();
+        REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
+    }
+
+    pub struct BarrierHandle {
+        key: (String, String),
+        arrived: Arc<Notify>,
+        release: Arc<Notify>,
+    }
+
+    pub fn arm(bucket: &str, object: &str, part_index: usize) -> BarrierHandle {
+        let key = (bucket.to_string(), object.to_string());
+        let arrived = Arc::new(Notify::new());
+        let release = Arc::new(Notify::new());
+        let previous = registry().lock().expect("GET part barrier registry poisoned").insert(
+            key.clone(),
+            Armed {
+                part_index,
+                arrived: Arc::clone(&arrived),
+                release: Arc::clone(&release),
+            },
+        );
+        assert!(previous.is_none(), "GET part barrier already armed for object");
+        BarrierHandle { key, arrived, release }
+    }
+
+    impl BarrierHandle {
+        pub async fn wait_until_paused(&self) {
+            self.arrived.notified().await;
+        }
+
+        pub fn release(&self) {
+            self.release.notify_one();
+        }
+    }
+
+    impl Drop for BarrierHandle {
+        fn drop(&mut self) {
+            self.release.notify_one();
+            registry()
+                .lock()
+                .expect("GET part barrier registry poisoned")
+                .remove(&self.key);
+        }
+    }
+
+    pub async fn checkpoint(bucket: &str, object: &str, part_index: usize) {
+        let state = registry()
+            .lock()
+            .expect("GET part barrier registry poisoned")
+            .get(&(bucket.to_string(), object.to_string()))
+            .filter(|armed| armed.part_index == part_index)
+            .map(|armed| (Arc::clone(&armed.arrived), Arc::clone(&armed.release)));
+        if let Some((arrived, release)) = state {
+            arrived.notify_one();
+            release.notified().await;
+        }
+    }
+}
+
 #[cfg(test)]
 mod metadata_cache_tests {
     use super::*;

crates/ecstore/src/store/object.rs

@@ -16,7 +16,7 @@ use super::*;
 use crate::disk::OldCurrentSize;
 use crate::set_disk::{
     get_lock_acquire_timeout, get_object_lock_diag_slow_acquire_threshold, get_object_lock_diag_slow_hold_threshold,
-    is_lock_optimization_enabled, is_object_lock_diag_enabled,
+    is_object_lock_diag_enabled,
 };
 use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
 use rustfs_io_metrics::{
@@ -89,6 +89,7 @@ impl PreparedGetObjectReader {
 struct LockGuardedReader {
     inner: Box<dyn AsyncRead + Unpin + Send + Sync>,
     guard: Option<ObjectLockDiagGuard>,
+    remaining: Option<usize>,
 }
 
 impl AsyncRead for LockGuardedReader {
@@ -96,8 +97,23 @@ impl AsyncRead for LockGuardedReader {
         let had_capacity = buf.remaining() > 0;
         let filled_before = buf.filled().len();
         let poll = Pin::new(&mut self.inner).poll_read(cx, buf);
-        if had_capacity && matches!(poll, Poll::Ready(Ok(()))) && buf.filled().len() == filled_before {
-            self.guard.take();
+        match &poll {
+            Poll::Ready(Ok(())) if buf.filled().len() > filled_before => {
+                let produced = buf.filled().len() - filled_before;
+                if let Some(remaining) = self.remaining.as_mut() {
+                    *remaining = remaining.saturating_sub(produced);
+                    if *remaining == 0 {
+                        self.guard.take();
+                    }
+                }
+            }
+            Poll::Ready(Ok(())) if had_capacity => {
+                self.guard.take();
+            }
+            Poll::Ready(Err(_)) => {
+                self.guard.take();
+            }
+            _ => {}
         }
         poll
     }
@@ -641,14 +657,16 @@ impl ECStore {
     }
 
     fn attach_read_lock_guard(mut reader: GetObjectReader, guard: Option<ObjectLockDiagGuard>) -> GetObjectReader {
-        if is_lock_optimization_enabled() || reader.buffered_body.is_some() {
+        if reader.buffered_body.is_some() || reader.object_info.size == 0 {
             return reader;
         }
 
         if let Some(guard) = guard {
+            let remaining = usize::try_from(reader.object_info.size).ok();
             reader.stream = Box::new(LockGuardedReader {
                 inner: reader.stream,
                 guard: Some(guard),
+                remaining,
             });
         }
 
@@ -2649,8 +2667,11 @@ mod tests {
                 ObjectLockDiagMode::Read,
             );
             let reader = GetObjectReader {
-                stream: Box::new(Cursor::new(Vec::<u8>::new())),
-                object_info: ObjectInfo::default(),
+                stream: Box::new(Cursor::new(vec![1])),
+                object_info: ObjectInfo {
+                    size: 1,
+                    ..Default::default()
+                },
                 buffered_body: None,
                 body_source: Default::default(),
             };
@@ -2670,7 +2691,7 @@ mod tests {
 
     #[tokio::test]
     #[serial_test::serial]
-    async fn reader_lock_is_not_held_for_stream_when_optimization_is_enabled() {
+    async fn reader_lock_is_held_for_stream_when_optimization_is_enabled() {
         temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, Some("true"))], async {
             let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
             let lock = rustfs_lock::NamespaceLock::with_local_manager("test".to_string(), manager);
@@ -2690,17 +2711,23 @@ mod tests {
             );
             let reader = GetObjectReader {
                 stream: Box::new(Cursor::new(vec![1, 2, 3])),
-                object_info: ObjectInfo::default(),
+                object_info: ObjectInfo {
+                    size: 3,
+                    ..Default::default()
+                },
                 buffered_body: None,
                 body_source: Default::default(),
             };
 
             let reader = ECStore::attach_read_lock_guard(reader, Some(read_guard));
 
+            lock.get_write_lock(key.clone(), "writer", Duration::from_millis(20))
+                .await
+                .expect_err("streaming readers must retain the read lock when lock optimization is enabled");
+            drop(reader);
             lock.get_write_lock(key, "writer", Duration::from_secs(1))
                 .await
-                .expect("lock optimization should release the read lock before returning the stream");
-            drop(reader);
+                .expect("cancelling the reader should release the read lock");
         })
         .await;
     }
@@ -2744,41 +2771,50 @@ mod tests {
 
     #[tokio::test]
     #[serial_test::serial]
-    async fn reader_lock_is_released_after_stream_eof() {
-        temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, Some("false"))], async {
-            let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
-            let lock = rustfs_lock::NamespaceLock::with_local_manager("test".to_string(), manager);
-            let key = rustfs_lock::ObjectKey::new("bucket", "object");
-            let read_guard = lock
-                .get_read_lock(key.clone(), "reader", Duration::from_secs(1))
-                .await
-                .expect("read lock should be acquired");
-            let read_guard = ObjectLockDiagGuard::new(
-                read_guard,
-                true,
-                "test_get_object",
-                Some("bucket".to_string()),
-                Some("object".to_string()),
-                Some("reader".to_string()),
-                ObjectLockDiagMode::Read,
-            );
-            let reader = GetObjectReader {
-                stream: Box::new(Cursor::new(vec![1, 2, 3])),
-                object_info: ObjectInfo::default(),
-                buffered_body: None,
-                body_source: Default::default(),
-            };
+    async fn reader_lock_is_released_after_exact_content_length_without_extra_eof_poll() {
+        for enabled in ["false", "true"] {
+            temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, Some(enabled))], async {
+                let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+                let lock = rustfs_lock::NamespaceLock::with_local_manager("test".to_string(), manager);
+                let key = rustfs_lock::ObjectKey::new("bucket", "object");
+                let read_guard = lock
+                    .get_read_lock(key.clone(), "reader", Duration::from_secs(1))
+                    .await
+                    .expect("read lock should be acquired");
+                let read_guard = ObjectLockDiagGuard::new(
+                    read_guard,
+                    true,
+                    "test_get_object",
+                    Some("bucket".to_string()),
+                    Some("object".to_string()),
+                    Some("reader".to_string()),
+                    ObjectLockDiagMode::Read,
+                );
+                let reader = GetObjectReader {
+                    stream: Box::new(Cursor::new(vec![1, 2, 3])),
+                    object_info: ObjectInfo {
+                        size: 3,
+                        ..Default::default()
+                    },
+                    buffered_body: None,
+                    body_source: Default::default(),
+                };
 
-            let mut reader = ECStore::attach_read_lock_guard(reader, Some(read_guard));
-            let mut output = Vec::new();
-            reader.stream.read_to_end(&mut output).await.expect("reader should reach EOF");
-            assert_eq!(output, vec![1, 2, 3]);
+                let mut reader = ECStore::attach_read_lock_guard(reader, Some(read_guard));
+                let mut output = [0_u8; 3];
+                reader
+                    .stream
+                    .read_exact(&mut output)
+                    .await
+                    .expect("reader should deliver the exact content length");
+                assert_eq!(output, [1, 2, 3]);
 
-            lock.get_write_lock(key, "writer", Duration::from_secs(1))
-                .await
-                .expect("EOF should release the read lock before the reader is dropped");
-            drop(reader);
-        })
-        .await;
+                lock.get_write_lock(key, "writer", Duration::from_secs(1))
+                    .await
+                    .expect("the final response byte should release the read lock without an extra EOF poll");
+                drop(reader);
+            })
+            .await;
+        }
     }
 }