fix(storage): bound conditional file lock artifacts

crates/ecstore/src/disk/local.rs

@@ -7949,10 +7949,15 @@ impl DiskAPI for LocalDisk {
             use std::io::Write as _;
 
             let file_path = self.io_get_object_path(volume, path)?;
-            let lock_path = file_path.with_extension("rustfs-cas.lock");
             let path = path.to_string();
             let sync_metadata = effective_durability(volume).syncs_commit_metadata();
             return Ok(tokio::task::spawn_blocking(move || {
+                // A persistent directory lock bounds metadata growth. Removing
+                // per-target lock files can split flock ownership across inodes.
+                let lock_path = file_path
+                    .parent()
+                    .ok_or_else(|| std::io::Error::new(ErrorKind::InvalidInput, "conditional file has no parent"))?
+                    .join(".rustfs-cas.lock");
                 let lock = std::fs::OpenOptions::new()
                     .create(true)
                     .truncate(false)
@@ -21841,7 +21846,10 @@ mod test {
         let marker_path = disk
             .get_object_path(RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
             .expect("marker path should resolve");
-        let lock_path = marker_path.with_extension("rustfs-cas.lock");
+        let lock_path = marker_path
+            .parent()
+            .expect("marker path should have a parent")
+            .join(".rustfs-cas.lock");
         let lock = std::fs::OpenOptions::new()
             .create(true)
             .truncate(false)

crates/heal/src/heal/resume/tests.rs

@@ -1842,6 +1842,27 @@ async fn deleted_checkpoint_is_not_recreated_by_an_old_manager() {
     temp_dir.close().expect("remove deleted checkpoint test directory");
 }
 
+#[cfg(unix)]
+#[tokio::test]
+async fn checkpoint_cleanup_leaves_no_task_specific_lock_artifact() {
+    let (temp_dir, disk) = schema_test_disk().await;
+    let task_id = ResumeUtils::generate_task_id();
+    let manager = CheckpointManager::new(disk.clone(), task_id.clone())
+        .await
+        .expect("create checkpoint manager");
+    let lock_path = Path::new(BUCKET_META_PREFIX)
+        .join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}"))
+        .with_extension("rustfs-cas.lock");
+    let lock_path = temp_dir.path().join(RUSTFS_META_BUCKET).join(lock_path);
+
+    manager.cleanup().await.expect("delete checkpoint fixture");
+
+    assert!(
+        !lock_path.exists(),
+        "successful checkpoint cleanup must not leave a task-specific lock artifact"
+    );
+}
+
 #[tokio::test]
 async fn an_empty_blocked_marker_still_blocks_resume_selection() {
     let (temp_dir, disk) = schema_test_disk().await;