Have scheduler LSF driver not bhist jobs newer than POLL_PERIOD

This commit makes lsf driver not bhist jobs submitted after previous polling. This fixes the issue where we would bhist for jobs recently submitted, because sometimes LSF takes a couple of milliseconds to process a submission before answering for it in bjobs.

src/ert/scheduler/lsf_driver.py

@@ -21,7 +21,6 @@
     MutableMapping,
     Optional,
     Sequence,
-    Tuple,
     Union,
     get_args,
 )
@@ -86,6 +85,12 @@ class _Stat(BaseModel):
     jobs: Mapping[str, AnyJob]
 
 
+class JobData(BaseModel):
+    iens: int
+    job_state: AnyJob
+    submitted_timestamp: float
+
+
 def parse_bjobs(bjobs_output: str) -> Dict[str, Dict[str, Dict[str, str]]]:
     data: Dict[str, Dict[str, str]] = {}
     for line in bjobs_output.splitlines():
@@ -166,6 +171,16 @@ def parse_bhist(bhist_output: str) -> Dict[str, Dict[str, int]]:
     return data
 
 
+def filter_job_ids_on_submission_time(
+    jobs: MutableMapping[str, JobData], submitted_before: float
+) -> set[str]:
+    return {
+        job_id
+        for job_id, job_data in jobs.items()
+        if submitted_before > job_data.submitted_timestamp
+    }
+
+
 class LsfDriver(Driver):
     def __init__(
         self,
@@ -189,7 +204,7 @@ def __init__(
         self._bjobs_cmd = Path(bjobs_cmd or shutil.which("bjobs") or "bjobs")
         self._bkill_cmd = Path(bkill_cmd or shutil.which("bkill") or "bkill")
 
-        self._jobs: MutableMapping[str, Tuple[int, AnyJob]] = {}
+        self._jobs: MutableMapping[str, JobData] = {}
         self._iens2jobid: MutableMapping[int, str] = {}
         self._max_attempt: int = 100
         self._sleep_time_between_bkills = 30
@@ -261,7 +276,11 @@ async def submit(
         (Path(runpath) / LSF_INFO_JSON_FILENAME).write_text(
             json.dumps({"job_id": job_id}), encoding="utf-8"
         )
-        self._jobs[job_id] = (iens, QueuedJob(job_state="PEND"))
+        self._jobs[job_id] = JobData(
+            iens=iens,
+            job_state=QueuedJob(job_state="PEND"),
+            submitted_timestamp=time.time(),
+        )
         self._iens2jobid[iens] = job_id
 
     async def kill(self, iens: int) -> None:
@@ -319,8 +338,12 @@ async def poll(self) -> None:
                 )
             bjobs_states = _Stat(**parse_bjobs(stdout.decode(errors="ignore")))
 
-            if missing_in_bjobs_output := set(current_jobids) - set(
-                bjobs_states.jobs.keys()
+            job_ids_found_in_bjobs_output = set(bjobs_states.jobs.keys())
+            if (
+                missing_in_bjobs_output := filter_job_ids_on_submission_time(
+                    self._jobs, submitted_before=time.time() - self._poll_period
+                )
+                - job_ids_found_in_bjobs_output
             ):
                 logger.debug(f"bhist is used for job ids: {missing_in_bjobs_output}")
                 bhist_states = await self._poll_once_by_bhist(missing_in_bjobs_output)
@@ -346,7 +369,8 @@ async def _process_job_update(self, job_id: str, new_state: AnyJob) -> None:
         if job_id not in self._jobs:
             return
 
-        iens, old_state = self._jobs[job_id]
+        old_state = self._jobs[job_id].job_state
+        iens = self._jobs[job_id].iens
         if isinstance(new_state, IgnoredJobstates):
             logger.debug(
                 f"Job ID '{job_id}' for {iens=} is of unknown job state '{new_state.job_state}'"
@@ -356,7 +380,7 @@ async def _process_job_update(self, job_id: str, new_state: AnyJob) -> None:
         if _STATE_ORDER[type(new_state)] <= _STATE_ORDER[type(old_state)]:
             return
 
-        self._jobs[job_id] = (iens, new_state)
+        self._jobs[job_id].job_state = new_state
         event: Optional[Event] = None
         if isinstance(new_state, RunningJob):
             logger.debug(f"Realization {iens} is running")

tests/unit_tests/scheduler/test_lsf_driver.py

@@ -19,12 +19,14 @@
     FinishedEvent,
     FinishedJobFailure,
     FinishedJobSuccess,
+    JobData,
     JobState,
     QueuedJob,
     RunningJob,
     StartedEvent,
     _Stat,
     build_resource_requirement_string,
+    filter_job_ids_on_submission_time,
     parse_bhist,
     parse_bjobs,
 )
@@ -77,7 +79,11 @@ async def test_events_produced_from_jobstate_updates(jobstate_sequence: List[str
 
     async def mocked_submit(self, iens, *_args, **_kwargs):
         """A mocked submit is speedier than going through a command on disk"""
-        self._jobs["1"] = (iens, QueuedJob(job_state="PEND"))
+        self._jobs["1"] = JobData(
+            iens=iens,
+            job_state=QueuedJob(job_state="PEND"),
+            submitted_timestamp=time.time(),
+        )
         self._iens2jobid[iens] = "1"
 
     driver.submit = mocked_submit.__get__(driver)
@@ -95,14 +101,14 @@ async def mocked_submit(self, iens, *_args, **_kwargs):
     if not started and not finished_success and not finished_failure:
         assert len(events) == 0
 
-        iens, state = driver._jobs["1"]
+        iens, state = driver._jobs["1"].iens, driver._jobs["1"].job_state
         assert iens == 0
         assert isinstance(state, QueuedJob)
     elif started and not finished_success and not finished_failure:
         assert len(events) == 1
         assert events[0] == StartedEvent(iens=0)
 
-        iens, state = driver._jobs["1"]
+        iens, state = driver._jobs["1"].iens, driver._jobs["1"].job_state
         assert iens == 0
         assert isinstance(state, RunningJob)
     elif started and finished_success and finished_failure:
@@ -727,3 +733,28 @@ async def mock_submit(*args, **kwargs):
     assert "LSF kill failed" not in caplog.text
     assert "LSF kill failed" not in capsys.readouterr().err
     assert "LSF kill failed" not in capsys.readouterr().out
+
+
+@pytest.mark.parametrize(
+    "time_submitted_modifier, expected_result",
+    [
+        pytest.param(
+            -1.0,
+            set(["1"]),
+            id="job_submitted_before_deadline",
+        ),
+        pytest.param(0, set(), id="job_submitted_on_deadline"),
+        pytest.param(1.0, set(), id="job_submitted_after_deadline"),
+    ],
+)
+def test_filter_job_ids_on_submission_time(time_submitted_modifier, expected_result):
+    submitted_before = time.time()
+    job_submitted_timestamp = submitted_before + time_submitted_modifier
+    jobs = {
+        "1": JobData(
+            iens=0,
+            job_state=QueuedJob(job_state="PEND"),
+            submitted_timestamp=job_submitted_timestamp,
+        )
+    }
+    assert filter_job_ids_on_submission_time(jobs, submitted_before) == expected_result