fix: in remote jobs, upload storage in topological order such that modification dates are preserved (e.g. in case of group jobs) (#3377)

johanneskoester · tadasbare · web-flow · commit eace08f356b3 · 2025-03-11T16:10:16.000+01:00
### Description

&lt;!--Add a description of your PR here--&gt;

### QC
&lt;!-- Make sure that you can tick the boxes below. --&gt;

* [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).


&lt;!-- This is an auto-generated comment: release notes by coderabbit.ai
--&gt;
## Summary by CodeRabbit

- **Refactor**
- Improved the job output storage workflow by changing the data
structure for job processing.

- **New Features**
- Included template files in the package distribution for enhanced
functionality.

- **Chores**
- Restructured dependencies in the test environment to treat certain
tools as top-level dependencies, streamlining package management.
&lt;!-- end of auto-generated comment: release notes by coderabbit.ai --&gt;

---------

Co-authored-by: Tadas Bareikis &lt;tadas@vugene.com&gt;
diff --git a/snakemake/dag.py b/snakemake/dag.py
@@ -409,25 +409,28 @@ async def store_storage_outputs(self):
         if self.workflow.remote_exec:
             logger.info("Storing output in storage.")
             try:
-                async with asyncio.TaskGroup() as tg:
-                    for job in self.needrun_jobs(exclude_finished=False):
-                        benchmark = [job.benchmark] if job.benchmark else []
-
-                        async def tostore(f):
-                            return (
-                                f.is_storage
-                                and f
-                                not in self.workflow.storage_settings.unneeded_temp_files
-                                and await f.exists_local()
-                            )
+                for level in self.toposorted(
+                    set(self.needrun_jobs(exclude_finished=False))
+                ):
+                    async with asyncio.TaskGroup() as tg:
+                        for job in level:
+                            benchmark = [job.benchmark] if job.benchmark else []
+
+                            async def tostore(f):
+                                return (
+                                    f.is_storage
+                                    and f
+                                    not in self.workflow.storage_settings.unneeded_temp_files
+                                    and await f.exists_local()
+                                )
 
-                        if self.finished(job):
-                            for f in chain(job.output, benchmark):
+                            if self.finished(job):
+                                for f in chain(job.output, benchmark):
+                                    if await tostore(f):
+                                        tg.create_task(f.store_in_storage())
+                            for f in job.log:
                                 if await tostore(f):
                                     tg.create_task(f.store_in_storage())
-                        for f in job.log:
-                            if await tostore(f):
-                                tg.create_task(f.store_in_storage())
             except ExceptionGroup as e:
                 raise WorkflowError("Failed to store output in storage.", e)
 
