fix: properly delete local copies of storage files after remote jobs (#…

…2793)

### Description

<!--Add a description of your PR here-->

### QC
<!-- Make sure that you can tick the boxes below. -->

* [x] The PR contains a test case for the changes or the changes are
already covered by an existing test case.
* [x] The documentation (`docs/`) is updated to reflect the changes or
this is not necessary (e.g. if the change does neither modify the
language nor the behavior or functionalities of Snakemake).

snakemake/dag.py

@@ -410,6 +410,7 @@ def cleanup_storage_objects(self):
                 for f in chain(job.input, job.output):
                     if f.is_storage and f not in cleaned:
                         f.storage_object.cleanup()
+                        f.remove(only_local=True)
                         cleaned.add(f)
 
     def create_conda_envs(self, dryrun=False, quiet=False):

snakemake/io.py

@@ -667,11 +667,13 @@ def protect(self):
         # protect explicit output itself
         lchmod(self.file, mode)
 
-    async def remove(self, remove_non_empty_dir=False):
+    async def remove(self, remove_non_empty_dir=False, only_local=False):
         if self.is_directory:
-            await remove(self, remove_non_empty_dir=True)
+            await remove(self, remove_non_empty_dir=True, only_local=only_local)
         else:
-            await remove(self, remove_non_empty_dir=remove_non_empty_dir)
+            await remove(
+                self, remove_non_empty_dir=remove_non_empty_dir, only_local=only_local
+            )
 
     def touch(self, times=None):
         """times must be 2-tuple: (atime, mtime)"""
@@ -947,8 +949,8 @@ def contains_wildcard_constraints(pattern):
     return any(match.group("constraint") for match in WILDCARD_REGEX.finditer(pattern))
 
 
-async def remove(file, remove_non_empty_dir=False):
-    if file.is_storage and file.should_not_be_retrieved_from_storage:
+async def remove(file, remove_non_empty_dir=False, only_local=False):
+    if not only_local and file.is_storage and file.should_not_be_retrieved_from_storage:
         if await file.exists_in_storage():
             await file.storage_object.managed_remove()
     elif os.path.isdir(file) and not os.path.islink(file):