Fix job ID missing from DB/data store for submit-failed jobs

Author: MetRonnie

cylc/flow/data_store_mgr.py

@@ -67,7 +67,6 @@
     Set,
     TYPE_CHECKING,
     Tuple,
-    Union,
 )
 import zlib
 
@@ -1612,37 +1611,27 @@ def _apply_broadcasts_to_runtime(self, tokens, rtconfig):
 
     def insert_job(
         self,
-        name: str,
-        cycle_point: Union['PointBase', str],
+        itask: 'TaskProxy',
         status: str,
         job_conf: dict,
-    ):
+    ) -> None:
         """Insert job into data-store.
 
         Args:
-            name: Corresponding task name.
-            cycle_point: Cycle point string
             status: The task's state.
             job_conf:
                 Dictionary of job configuration used to generate
                 the job script.
                 (see TaskJobManager._prep_submit_task_job_impl)
 
-        Returns:
-
-            None
-
         """
         if status not in JOB_STATUS_SET:
             # Ignore task-only states e.g. preparing
             # https://github.com/cylc/cylc-flow/issues/4994
             return
 
         sub_num = job_conf['submit_num']
-        tp_tokens = self.id_.duplicate(
-            cycle=str(cycle_point),
-            task=name,
-        )
+        tp_tokens = self.id_.duplicate(itask.tokens)
         tproxy: Optional[PbTaskProxy]
         tp_id, tproxy = self.store_node_fetcher(tp_tokens)
         if not tproxy:
@@ -1665,6 +1654,7 @@ def insert_job(
             execution_time_limit=job_conf.get('execution_time_limit'),
             platform=job_conf['platform']['name'],
             job_runner_name=job_conf.get('job_runner_name'),
+            job_id=itask.summary.get('submit_method_id'),
         )
         # Not all fields are populated with some submit-failures,
         # so use task cfg as base.

cylc/flow/task_events_mgr.py

@@ -794,13 +794,7 @@ def process_message(
             self._process_message_submitted(itask, event_time, forced)
             self.spawn_children(itask, TASK_OUTPUT_SUBMITTED, forced)
 
-            # ... but either way update the job ID in the job proxy (it only
-            # comes in via the submission message).
-            if itask.run_mode != RunMode.SIMULATION:
-                self.data_store_mgr.delta_job_attr(
-                    itask, 'job_id', itask.summary['submit_method_id']
-                )
-            else:
+            if itask.run_mode == RunMode.SIMULATION:
                 # In simulation mode submitted implies started:
                 self.spawn_children(itask, TASK_OUTPUT_STARTED, forced)
 
@@ -1465,7 +1459,6 @@ def _process_message_submit_failed(
             "time_submit_exit": event_time,
             "submit_status": 1,
         })
-        itask.summary['submit_method_id'] = None
         LOG.error(f"[{itask}] {self.EVENT_SUBMIT_FAILED}")
         if (
             forced
@@ -1590,11 +1583,9 @@ def _insert_task_job(
             except IndexError:
                 # we do not have access to the job config (e.g. Scheduler
                 # crashed) - https://github.com/cylc/cylc-flow/pull/6326
-                job_id = itask.tokens.duplicate(
-                    job=itask.submit_num
-                ).relative_id
                 LOG.warning(
-                    f'Could not find the job configuration for "{job_id}".'
+                    'Could not find the job configuration for '
+                    f'"{itask.job_tokens.relative_id}".'
                 )
                 itask.jobs.append({"submit_num": itask.submit_num})
                 job_conf = itask.jobs[-1]
@@ -1611,8 +1602,7 @@ def _insert_task_job(
 
         # insert job into data store
         self.data_store_mgr.insert_job(
-            itask.tdef.name,
-            itask.point,
+            itask,
             job_status,
             {
                 **job_conf,
@@ -1635,7 +1625,7 @@ def _insert_task_job(
                 # preparation started due to intelligent host (and or
                 # platform) selection
                 'platform_name': itask.platform['name'],
-            }
+            },
         )
 
     def _setup_job_logs_retrieval(self, itask, event) -> None:

tests/integration/network/test_graphql.py

@@ -366,10 +366,12 @@ async def test_edges(harness):
 
 
 async def test_jobs(harness):
+    schd: Scheduler
     schd, client, w_tokens = harness
 
     # add a job
-    schd.data_store_mgr.insert_job('a', '1', 'submitted', job_config(schd))
+    itask = schd.pool._get_task_by_id('1/a')
+    schd.data_store_mgr.insert_job(itask, 'submitted', job_config(schd))
     schd.data_store_mgr.update_data_structure()
     j_tokens = w_tokens.duplicate(
         cycle='1',

tests/integration/test_data_store_mgr.py

@@ -316,15 +316,18 @@ async def test_delta_task_held(mod_harness):
 
 def test_insert_job(mod_harness):
     """Test method that adds a new job to the store."""
+    schd: Scheduler
     schd, data = mod_harness
     assert len(schd.data_store_mgr.added[JOBS]) == 0
-    schd.data_store_mgr.insert_job('foo', '1', 'submitted', job_config(schd))
+    itask = schd.pool.get_tasks()[0]
+    schd.data_store_mgr.insert_job(itask, 'submitted', job_config(schd))
     assert len(schd.data_store_mgr.added[JOBS]) == 1
     assert ext_id(schd) in schd.data_store_mgr.added[JOBS]
 
 
 def test_insert_db_job(mod_harness, job_db_row):
     """Test method that adds a new job from the db to the store."""
+    schd: Scheduler
     schd, data = mod_harness
     assert len(schd.data_store_mgr.added[JOBS]) == 1
     schd.data_store_mgr.insert_db_job(0, job_db_row)
@@ -334,6 +337,7 @@ def test_insert_db_job(mod_harness, job_db_row):
 
 def test_delta_job_msg(mod_harness):
     """Test method adding messages to job element."""
+    schd: Scheduler
     schd, data = mod_harness
     j_id = ext_id(schd)
     tokens = Tokens(j_id)
@@ -724,7 +728,7 @@ def _patch_remove(*args, **kwargs):
         assert itask
         itask.submit_num += 1
         one.data_store_mgr.insert_job(
-            itask.tdef.name, itask.point, itask.state.status, {'submit_num': 1}
+            itask, itask.state.status, {'submit_num': 1}
         )
         await one.update_data_structure()
 

tests/integration/test_task_events_mgr.py

@@ -31,6 +31,10 @@
     EventKey,
     TaskJobLogsRetrieveContext,
 )
+from cylc.flow.task_state import (
+    TASK_STATUS_PREPARING,
+    TASK_STATUS_SUBMIT_FAILED,
+)
 
 from .test_workflow_events import TEMPLATES
 
@@ -126,7 +130,7 @@ async def test__insert_task_job(flow, one_conf, scheduler, start, validate):
 
 
 async def test__always_insert_task_job(
-    flow, scheduler, mock_glbl_cfg, start, run
+    flow, scheduler, mock_glbl_cfg, start
 ):
     """Insert Task Job _Always_ inserts a task into the data store.
 
@@ -189,6 +193,46 @@ async def test__always_insert_task_job(
             assert job.submitted_time
 
 
+async def test__submit_failed_job_id(flow, scheduler, start, db_select):
+    """If a job is killed in the submitted state, the job ID should still be
+    in the DB/data store.
+
+    See https://github.com/cylc/cylc-flow/pull/6926
+    """
+    async def get_ds_job_id(schd: Scheduler):
+        await schd.update_data_structure()
+        return list(schd.data_store_mgr.data[schd.id][JOBS].values())[0].job_id
+
+    id_ = flow('foo')
+    schd: Scheduler = scheduler(id_)
+    job_id = '1234'
+    async with start(schd):
+        itask = schd.pool.get_tasks()[0]
+        itask.state_reset(TASK_STATUS_PREPARING)
+        itask.submit_num = 1
+        itask.summary['submit_method_id'] = job_id
+        schd.workflow_db_mgr.put_insert_task_jobs(itask, {})
+        schd.task_events_mgr.process_message(
+            itask, 'INFO', schd.task_events_mgr.EVENT_SUBMITTED
+        )
+        assert await get_ds_job_id(schd) == job_id
+
+        schd.task_events_mgr.process_message(
+            itask, 'CRITICAL', schd.task_events_mgr.EVENT_SUBMIT_FAILED
+        )
+        assert itask.state(TASK_STATUS_SUBMIT_FAILED)
+        assert await get_ds_job_id(schd) == job_id
+
+    assert db_select(schd, False, 'task_jobs', 'job_id', 'submit_status') == [
+        (job_id, 1)
+    ]
+
+    # Restart and check data store again:
+    schd = scheduler(id_)
+    async with start(schd):
+        assert await get_ds_job_id(schd) == job_id
+
+
 async def test__process_message_failed_with_retry(one, start, log_filter):
     """Log job failure, even if a retry is scheduled.
 

tests/integration/tui/test_logs.py

@@ -149,8 +149,7 @@ async def workflow(
         # mark 1/a/01 as failed
         job_1 = schd.tokens.duplicate(cycle='1', task='a', job='01')
         schd.data_store_mgr.insert_job(
-            'a',
-            IntegerPoint('1'),
+            itask,
             TASK_STATUS_SUCCEEDED,
             {'submit_num': 1, 'platform': {'name': 'x'}}
         )
@@ -160,8 +159,7 @@ async def workflow(
         itask.submit_num = 2
         job_2 = schd.tokens.duplicate(cycle='1', task='a', job='02')
         schd.data_store_mgr.insert_job(
-            'a',
-            IntegerPoint('1'),
+            itask,
             TASK_STATUS_SUCCEEDED,
             {'submit_num': 2, 'platform': {'name': 'x'}}
         )