Fix job runner missing from data store for submit-failed jobs

Remove spurious submit-failed dummy job in task proxy job list

Author: MetRonnie

cylc/flow/task_job_mgr.py

@@ -1286,6 +1286,12 @@ def _prep_submit_task_job(
                     # bad_hosts:
                     self.bad_hosts -= exc.hosts_consumed
                     self._set_retry_timers(itask, rtconfig)
+                # Provide dummy platform otherwise it will incorrectly show as
+                # the default localhost platform in the data store:
+                itask.platform = {
+                    'name': rtconfig['platform'],
+                    'job runner': '',
+                }
                 self._prep_submit_task_job_error(itask, msg, exc)
                 return False
 
@@ -1337,15 +1343,20 @@ def _prep_submit_task_job_error(
         itask.is_manual_submit = False
         # job failed in preparation i.e. is really preparation-failed rather
         # than submit-failed
-        # provide a dummy job config - this info will be added to the data
-        # store
         try_num = itask.get_try_num()
-        itask.jobs.append({
-            'task_id': itask.identity,
-            'platform': itask.platform,
-            'submit_num': itask.submit_num,
-            'try_num': try_num,
-        })
+        if not itask.jobs or (
+            itask.jobs[-1]['submit_num'] != itask.submit_num
+        ):
+            # provide a dummy job config - this info will be added to the data
+            # store
+            itask.jobs.append({
+                'task_id': itask.identity,
+                'platform': itask.platform,
+                'job_runner_name': itask.platform['job runner'],
+                'submit_num': itask.submit_num,
+                'try_num': try_num,
+                'flow_nums': itask.flow_nums,
+            })
         # create a DB entry for the submit-failed job
         self.workflow_db_mgr.put_insert_task_jobs(
             itask,
@@ -1415,10 +1426,10 @@ def get_execution_time_limit(
 
     def get_job_conf(
         self,
-        itask,
-        rtconfig,
-        job_file_path=None,
-        job_d=None,
+        itask: 'TaskProxy',
+        rtconfig: dict,
+        job_file_path: Optional[str] = None,
+        job_d: Optional[str] = None,
     ):
         """Return a job config.
 

tests/integration/test_task_events_mgr.py

@@ -144,20 +144,22 @@ async def test__always_insert_task_job(
         [platforms]
             [[broken1]]
                 hosts = no-such-host-1
+                job runner = abc
             [[broken2]]
                 hosts = no-such-host-2
+                job runner = def
         [platform groups]
-            [[broken]]
+            [[broken_group]]
                 platforms = broken1
     """
     mock_glbl_cfg('cylc.flow.platforms.glbl_cfg', global_config)
 
     id_ = flow({
-        'scheduling': {'graph': {'R1': 'broken & broken2'}},
+        'scheduling': {'graph': {'R1': 'foo & bar'}},
         'runtime': {
             'root': {'submission retry delays': 'PT10M'},
-            'broken': {'platform': 'broken'},
-            'broken2': {'platform': 'broken2'}
+            'foo': {'platform': 'broken_group'},
+            'bar': {'platform': 'broken2'}
         }
     })
 
@@ -174,14 +176,17 @@ async def test__always_insert_task_job(
         )
 
         # Both jobs are in the data store with submit-failed state:
+        ds_jobs = schd.data_store_mgr.data[schd.id][JOBS]
         updates = {
-            k.split('//')[-1]: v.state
-            for k, v in schd.data_store_mgr.data[schd.id][JOBS].items()
+            id_.split('//')[-1]: (job.state, job.platform, job.job_runner_name)
+            for id_, job in ds_jobs.items()
         }
         assert updates == {
-            '1/broken/01': 'submit-failed',
-            '1/broken2/01': 'submit-failed'
+            '1/foo/01': ('submit-failed', 'broken_group', ''),
+            '1/bar/01': ('submit-failed', 'broken2', 'def'),
         }
+        for job in ds_jobs.values():
+            assert job.submitted_time
 
 
 async def test__process_message_failed_with_retry(one, start, log_filter):