fix: only download input for local jobs in the main process, not with…

…in remote groups (#2842)

### 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).

.github/workflows/main.yml

@@ -72,8 +72,8 @@ jobs:
 
           pip install -e .
 
-      - name: Configure fast APT mirror
-        uses: vegardit/fast-apt-mirror.sh@1.0.0
+      # - name: Configure fast APT mirror
+      #   uses: vegardit/fast-apt-mirror.sh@1.0.0
 
       - name: Setup apt dependencies
         run: |

snakemake/scheduler.py

@@ -298,11 +298,18 @@ def schedule(self):
                     local_runjobs = [job for job in run if job.is_local]
                     runjobs = [job for job in run if not job.is_local]
                     if local_runjobs:
-                        async_run(
-                            self.workflow.dag.retrieve_storage_inputs(
-                                jobs=local_runjobs, also_missing_internal=True
+                        if (
+                            not self.workflow.remote_exec
+                            and not self.workflow.local_exec
+                        ):
+                            # Workflow uses a remote plugin and this scheduling run
+                            # is on the main process. Hence, we have to download
+                            # non-shared remote files for the local jobs.
+                            async_run(
+                                self.workflow.dag.retrieve_storage_inputs(
+                                    jobs=local_runjobs, also_missing_internal=True
+                                )
                             )
-                        )
                         self.run(
                             local_runjobs,
                             executor=self._local_executor or self._executor,