Fix database issues (#2345)

* Fix database issues

* update test

Author: dbeltrankyl

autosubmit/history/database_managers/experiment_history_db_manager.py

@@ -403,10 +403,24 @@ def get_job_data_by_job_id_name(self, job_id: int, job_name: str) -> JobData:
         models = [Models.JobDataRow(*row) for row in job_data_rows][-1]
         return JobData.from_model(models)
 
-    def get_job_data_max_counter(self):
-        """ The max counter is the maximum count value for the count column in job_data. """
-        statement = "SELECT MAX(counter) as maxcounter FROM job_data"
-        counter_result = self.get_from_statement(self.historicaldb_file_path, statement)
+    def get_job_data_max_counter(self, job_name: str = None) -> int:
+        """
+        Get the maximum counter value from the `job_data` table. If a `job_name` is provided,
+        the query will filter by that specific job name.
+
+        :param job_name: The name of the job to filter by (optional).
+        :type job_name: str, optional
+        :return: The maximum counter value, or the default value if no rows are found.
+        :rtype: int
+        """
+        if job_name:
+            statement = "SELECT MAX(counter) as maxcounter FROM job_data WHERE job_name = ?"
+            arguments = (job_name,)
+            counter_result = self.get_from_statement_with_arguments(self.historicaldb_file_path, statement, arguments)
+        else:
+            statement = "SELECT MAX(counter) as maxcounter FROM job_data"
+            counter_result = self.get_from_statement(self.historicaldb_file_path, statement)
+
         if len(counter_result) <= 0:
             return DEFAULT_MAX_COUNTER
         else:
@@ -483,7 +497,7 @@ def _update_job_data_by_id(self, job_data_dc): ...
 
     def get_job_data_by_name(self, job_name): ...
 
-    def get_job_data_max_counter(self): ...
+    def get_job_data_max_counter(self, job_name: str = None) -> int: ...
 
     def delete_job_data(self, id_): ...
 

autosubmit/history/experiment_history.py

@@ -343,7 +343,7 @@ def _get_defined_queue_name(self, wrapper_queue, wrapper_code, qos):
     def _get_next_counter_by_job_name(self, job_name):
         """ Return the counter attribute from the latest job data row by job_name. """
         job_data_dc = self.manager.get_job_data_dc_unique_latest_by_job_name(job_name)
-        max_counter = self.manager.get_job_data_max_counter()
+        max_counter = self.manager.get_job_data_max_counter(job_name)
         if job_data_dc:
             return max(max_counter, job_data_dc.counter + 1)
         else:

autosubmit/job/job.py

@@ -1143,10 +1143,10 @@ def _get_from_stat(self, index: int, fail_count: int =-1) -> int:
             int: Value in the index position, or 0 if the file or index does not exist.
         """
         if fail_count == -1:
-            logname = os.path.join(self._log_path, f"{self.stat_file}0")
+            logname = os.path.join(self._tmp_path, f"{self.stat_file}0")
         else:
             fail_count = str(fail_count)
-            logname = os.path.join(self._log_path, f"{self.stat_file}{fail_count}")
+            logname = os.path.join(self._tmp_path, f"{self.stat_file}{fail_count}")
         if os.path.exists(logname):
             lines = open(logname).readlines()
             if len(lines) >= index + 1:
@@ -2549,10 +2549,12 @@ def write_end_time(self, completed, count=-1):
 
         end_time = self.check_end_time(count)
         if end_time > 0:
-            self.finish_time_timestamp = end_time
+            self.finish_time_timestamp = int(end_time)
+        if not self.finish_time_timestamp:
+            self.finish_time_timestamp = int(time.time())
         with open(Path(self._tmp_path) / f"{self.name}_TOTAL_STATS", 'a') as stat_file:
             stat_file.write(' ')
-            stat_file.write(date2str(datetime.datetime.fromtimestamp(float(self.finish_time_timestamp)), 'S'))
+            stat_file.write(date2str(datetime.datetime.fromtimestamp(int(self.finish_time_timestamp)), 'S'))
             stat_file.write(' ')
 
             if completed:

autosubmit/platforms/platform.py

@@ -1063,16 +1063,18 @@ def wait_until_timeout(self, timeout: int = 60) -> bool:
                 break
         return process_log
 
-    def recover_job_log(self, identifier: str, jobs_pending_to_process: Set[Any]) -> Set[Any]:
+    def recover_job_log(self, identifier: str, jobs_pending_to_process: Set[Any], as_conf: 'AutosubmitConfig') -> Set[Any]:
         """
-        Recovers log files for jobs from the recovery queue and retry failed jobs.
+        Recovers log files for jobs from the recovery queue and retries failed jobs.
 
-        Args:
-            identifier (str): Identifier for logging purposes.
-            jobs_pending_to_process (Set[Any]): Set of jobs that had issues during log retrieval.
-
-        Returns:
-            Set[Any]: Updated set of jobs pending to process.
+        :param identifier: Identifier for logging purposes.
+        :type identifier: str
+        :param jobs_pending_to_process: Set of jobs that had issues during log retrieval.
+        :type jobs_pending_to_process: Set[Any]
+        :param as_conf: The Autosubmit configuration object containing experiment data.
+        :type as_conf: AutosubmitConfig
+        :return: Updated set of jobs pending to process.
+        :rtype: Set[Any]
         """
         job = None
 
@@ -1093,7 +1095,7 @@ def recover_job_log(self, identifier: str, jobs_pending_to_process: Set[Any]) ->
                 pass
 
         if len(jobs_pending_to_process) > 0: # Restore the connection if there was an issue with one or more jobs.
-            self.restore_connection(None)
+            self.restore_connection(as_conf, log_recovery_process=True)
 
         # This second while is to keep retring the failed jobs.
         # With the unique queue, the main process won't send the job again, so we have to store it here.
@@ -1111,7 +1113,7 @@ def recover_job_log(self, identifier: str, jobs_pending_to_process: Set[Any]) ->
             Log.result(
                 f"{identifier} (Retry) Successfully recovered log for job '{job.name}' and retry '{job.fail_count}'.")
         if len(jobs_pending_to_process) > 0:
-            self.restore_connection(None)  # Restore the connection if there was an issue with one or more jobs.
+            self.restore_connection(as_conf, log_recovery_process=True)  # Restore the connection if there was an issue with one or more jobs.
 
         return jobs_pending_to_process
 
@@ -1132,9 +1134,9 @@ def recover_platform_job_logs(self, as_conf) -> None:
             # Keep alive signal timeout is 5 minutes, but the sleeptime is 60 seconds.
             self.keep_alive_timeout = max(log_recovery_timeout*5, 60*5)
             while self.wait_for_work(sleep_time=max(log_recovery_timeout, 60)):
-                jobs_pending_to_process = self.recover_job_log(identifier, jobs_pending_to_process)
+                jobs_pending_to_process = self.recover_job_log(identifier, jobs_pending_to_process, as_conf)
                 if self.cleanup_event.is_set():  # Check if the main process is waiting for this child to end.
-                    self.recover_job_log(identifier, jobs_pending_to_process)
+                    self.recover_job_log(identifier, jobs_pending_to_process, as_conf)
                     break
         except Exception as e:
             Log.error(f"{identifier} {e}")

test/unit/test_job.py

@@ -1903,14 +1903,14 @@ def test_get_from_stat(tmpdir, file_exists, index_timestamp, fail_count, expecte
 
     job = Job("dummy", 1, Status.WAITING, 0)
     assert job.stat_file == f"{job.name}_STAT_"
-    job._log_path = Path(tmpdir)
-    job._log_path.mkdir(parents=True, exist_ok=True)
+    job._tmp_path = Path(tmpdir)
+    job._tmp_path.mkdir(parents=True, exist_ok=True)
 
     # Generating the timestamp file
     if file_exists:
-        with open(job._log_path.joinpath(f"{job.stat_file}0"), "w") as stat_file:
+        with open(job._tmp_path.joinpath(f"{job.stat_file}0"), "w") as stat_file:
             stat_file.write("19704923\n19704924\n")
-        with open(job._log_path.joinpath(f"{job.stat_file}1"), "w") as stat_file:
+        with open(job._tmp_path.joinpath(f"{job.stat_file}1"), "w") as stat_file:
             stat_file.write("29704923\n29704924\n")
 
     if fail_count is None: