Recompute workflow resources until branch map is cached.

law/workflow/remote.py

@@ -551,22 +551,23 @@ def process_resources(self, force=False):
         task = self.task
 
         # collect resources over all branches if not just controlling running jobs
-        if (
-            not task.is_controlling_remote_jobs() and
-            (self._initial_process_resources is None or force)
-        ):
+        resources = self._initial_process_resources
+        if not task.is_controlling_remote_jobs() and (resources is None or force):
             get_job_resources = self._get_task_attribute("job_resources")
             branch_chunks = iter_chunks(task.branch_map.keys(), task.tasks_per_job)
-            self._initial_process_resources = {
+            resources = {
                 job_num: get_job_resources(job_num, branches)
                 for job_num, branches in enumerate(branch_chunks, 1)
                 if not self._can_skip_job(job_num, branches)
             }
+            # store resources only if the branch map was frozen
+            if task.cache_branch_map:
+                self._initial_process_resources = resources
 
-        if not self._initial_process_resources:
+        if not resources:
             return {}
 
-        return self._maximum_resources(self._initial_process_resources, self.poll_data.n_parallel)
+        return self._maximum_resources(resources, self.poll_data.n_parallel)
 
     def complete(self):
         if self.task.is_controlling_remote_jobs():
@@ -1039,7 +1040,7 @@ def poll(self):
 
         # the resources of yet unfinished jobs as claimed initially and reported to the scheduler
         # and the maximum amount resources potentially claimed by the jobs
-        job_resources = dict(self._initial_process_resources)
+        job_resources = dict(self._initial_process_resources or {})
         max_resources = self._maximum_resources(job_resources, self.poll_data.n_parallel)
 
         # track number of consecutive polling failures and the start time