fix: Convert Path to IOFile (snakemake#3405)

fgvieira · web-flow · commit c58684c67787 · 2025-03-14T10:51:20.000+01:00
Fixes snakemake#3193 Fixes snakemake#2933 ### QC  * [ ] The PR contains a test case for the changes or the changes are already covered by an existing test case. * [ ] 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).  ## Summary by CodeRabbit - **Refactor** - Revised handling of pending file operations to deliver more precise and actionable feedback. - Streamlined metadata cleanup processes to improve overall reliability and consistency. - Enhanced workflow routines for file cleanup, ensuring a more robust and efficient operation.
diff --git a/src/snakemake/dag.py b/src/snakemake/dag.py
@@ -499,12 +499,12 @@ async def check_incomplete(self):
         """Check if any output files are incomplete. This is done by looking up
         markers in the persistence module."""
         if not self.ignore_incomplete:
-            incomplete = await self.incomplete_files()
-            if incomplete:
+            incomplete_files = await self.incomplete_files()
+            if any(incomplete_files):
                 if self.workflow.dag_settings.force_incomplete:
-                    self.forcefiles.update(incomplete)
+                    self.forcefiles.update(incomplete_files)
                 else:
-                    raise IncompleteFilesException(incomplete)
+                    raise IncompleteFilesException(incomplete_files)
 
     def incomplete_external_jobid(self, job) -> Optional[str]:
         """Return the external jobid of the job if it is marked as incomplete.
@@ -599,10 +599,13 @@ async def incomplete_files(self):
         """Yield incomplete files."""
         incomplete = list()
         for job in filterfalse(self.needrun, self.jobs):
-            is_incomplete = await self.workflow.persistence.incomplete(job)
-            if is_incomplete:
-                for f in job.output:
-                    incomplete.append(f)
+            incomplete.extend(
+                [
+                    job
+                    for job in await self.workflow.persistence.incomplete(job)
+                    if job is not None
+                ]
+            )
         return incomplete
 
     @property
diff --git a/src/snakemake/persistence.py b/src/snakemake/persistence.py
@@ -228,7 +228,9 @@ def cleanup_locks(self):
         shutil.rmtree(self._lockdir)
 
     def cleanup_metadata(self, path):
-        return self._delete_record(self._metadata_path, path)
+        return self._delete_record(self._incomplete_path, path) or self._delete_record(
+            self._metadata_path, path
+        )
 
     def cleanup_shadow(self):
         if os.path.exists(self.shadow_path):
@@ -364,8 +366,7 @@ async def finished(self, job):
 
     def cleanup(self, job):
         for f in job.output:
-            self._delete_record(self._incomplete_path, f)
-            self._delete_record(self._metadata_path, f)
+            self.cleanup_metadata(f)
 
     async def incomplete(self, job):
         if self._incomplete_cache is None:
@@ -385,12 +386,12 @@ def marked_incomplete(f):
         async def is_incomplete(f):
             exists = await f.exists()
             marked = marked_incomplete(f)
-            return exists and marked
+            return f if exists and marked else None
 
         async with asyncio.TaskGroup() as tg:
             tasks = [tg.create_task(is_incomplete(f)) for f in job.output]
 
-        return any(task.result() for task in tasks)
+        return [task.result() for task in tasks]
 
     def _cache_incomplete_folder(self):
         self._incomplete_cache = {
diff --git a/src/snakemake/workflow.py b/src/snakemake/workflow.py
@@ -866,7 +866,7 @@ def cleanup_metadata(self, paths: List[Path]):
         )
         failed = []
         for path in paths:
-            success = self.persistence.cleanup_metadata(path)
+            success = self.persistence.cleanup_metadata(IOFile(path))
             if not success:
                 failed.append(str(path))
         if failed:

Original file line number	Diff line number	Diff line change
`@@ -866,7 +866,7 @@ def cleanup_metadata(self, paths: List[Path]):`
`866`	`866`	`)`
`867`	`867`	`failed = []`
`868`	`868`	`for path in paths:`
`869`		`- success = self.persistence.cleanup_metadata(path)`
	`869`	`+ success = self.persistence.cleanup_metadata(IOFile(path))`
`870`	`870`	`if not success:`
`871`	`871`	`failed.append(str(path))`
`872`	`872`	`if failed:`