Implements CSV rotation for job log files (#842)

This was originally an issue the team experienced once in a blue moon
where we got OSError because the csvs had become too large.

So this feature adds a mechanism to automatically rotate job log CSV
files (`initiated_jobs.csv` and `completed_jobs.csv`) once they exceed a
predefined maximum number of lines (default 10,000).

This prevents these files from growing indefinitely, which can improve
performance during file operations and better manage disk space. When
rotated, the existing file is renamed with a timestamp suffix, archiving
its content, and a new file is then implicitly created for ongoing logs.

Includes dedicated unit tests to ensure the rotation logic functions as
expected under various conditions. Also incorporates some minor test
data refactoring and housekeeping.

Author: calvinp0

arc/job/adapter.py

@@ -662,6 +662,32 @@ def determine_run_time(self):
         else:
             self.run_time = None
 
+    @staticmethod
+    def _rotate_csv_if_needed(csv_path: str, max_lines: int = 10000, line_count: Optional[int] = None) -> None:
+        """
+        Rotate a CSV file if it reaches or exceeds ``max_lines`` lines (including the header).
+        The archived file is renamed with a timestamp suffix in the same directory.
+        A fresh file (with headers) will be created by the caller as needed.
+
+        Args:
+            csv_path (str): The path to the CSV file.
+            max_lines (int): The maximum number of lines before rotation is triggered.
+            line_count (int, optional): Pre-computed line count to avoid re-reading the file.
+        """
+        if not os.path.isfile(csv_path):
+            return
+        if line_count is None:
+            with open(csv_path, 'r') as f:
+                line_count = sum(1 for _ in f)
+        if line_count >= max_lines:
+            local_time = datetime.datetime.now().strftime("%d_%m_%y")
+            base, ext = os.path.splitext(csv_path)
+            archive_path = f"{base}.old.{local_time}{ext}"
+            try:
+                os.rename(csv_path, archive_path)
+            except FileNotFoundError:
+                pass  # Another process already rotated the file.
+
     def _set_job_number(self):
         """
         Used as the entry number in the database, as well as the job name on the server.
@@ -686,12 +712,16 @@ def _set_job_number(self):
                 writer.writerow(row)
         with open(csv_path, 'r') as f:
             reader = csv.reader(f, dialect='excel')
+            line_count = 0
             job_num = 0
             for _ in reader:
+                line_count += 1
                 job_num += 1
                 if job_num == 100000:
                     job_num = 0
             self.job_num = job_num
+        # Rotate after counting to avoid a second full read of the file.
+        self._rotate_csv_if_needed(csv_path, max_lines=10000, line_count=line_count)
         # 2. Set other related attributes job_name and job_server_name.
         self.job_server_name = self.job_server_name or 'a' + str(self.job_num)
         if self.conformer is not None and self.job_name is None:
@@ -728,6 +758,7 @@ def write_completed_job_to_csv_file(self):
                 self.determine_job_status()
             local_arc_path_ = local_arc_path if os.path.isdir(local_arc_path) else ARC_PATH
             csv_path = os.path.join(local_arc_path_, 'completed_jobs.csv')
+            self._rotate_csv_if_needed(csv_path)
             if os.path.isfile(csv_path):
                 # check that this is the updated version
                 with open(csv_path, 'r') as f:

arc/job/adapter_test.py

@@ -6,8 +6,10 @@
 """
 
 import datetime
+import glob
 import math
 import os
+import tempfile
 import time
 import shutil
 import unittest
@@ -17,7 +19,7 @@
 
 from arc.common import ARC_TESTING_PATH
 from arc.imports import settings
-from arc.job.adapter import DataPoint, JobEnum, JobTypeEnum, JobExecutionTypeEnum
+from arc.job.adapter import DataPoint, JobAdapter, JobEnum, JobTypeEnum, JobExecutionTypeEnum
 from arc.job.adapters.gaussian import GaussianAdapter
 from arc.level import Level
 from arc.species import ARCSpecies
@@ -194,28 +196,34 @@ def setUpClass(cls):
                                     species=[ARCSpecies(label='spc1', xyz=['O 0 0 1'])],
                                     testing=True,
                                     )
+        # Copy the PBS time limit fixture into the directory structure the adapter expects.
+        stl_dir = os.path.join(ARC_TESTING_PATH, 'test_JobAdapter_ServerTimeLimit')
+        err_dest = os.path.join(stl_dir, 'calcs', 'Species', 'spc1', 'opt_101')
+        os.makedirs(err_dest, exist_ok=True)
+        shutil.copy(os.path.join(ARC_TESTING_PATH, 'server', 'pbs', 'timelimit', 'err.txt'),
+                    os.path.join(err_dest, 'err.txt'))
         cls.job_5 = GaussianAdapter(execution_type='queue',
-                                    job_name='spc1',
+                                    job_name='opt_101',
                                     job_type='opt',
                                     job_id='123456',
                                     job_num=101,
-                                    job_server_name = 'server3',
+                                    job_server_name='server3',
                                     level=Level(method='cbs-qb3'),
                                     project='test',
-                                    project_directory=os.path.join(ARC_TESTING_PATH, 'test_JobAdapter_ServerTimeLimit'),
+                                    project_directory=stl_dir,
                                     species=[ARCSpecies(label='spc1', xyz=['O 0 0 1'])],
                                     server='server3',
                                     testing=True,
                                     )
         cls.job_6 = GaussianAdapter(execution_type='queue',
-                                    job_name='spc1',
+                                    job_name='opt_101',
                                     job_type='opt',
                                     job_id='123456',
                                     job_num=101,
-                                    job_server_name = 'server1',
+                                    job_server_name='server1',
                                     level=Level(method='cbs-qb3'),
                                     project='test',
-                                    project_directory=os.path.join(ARC_TESTING_PATH, 'test_JobAdapter_ServerTimeLimit'),
+                                    project_directory=stl_dir,
                                     species=[ARCSpecies(label='spc1', xyz=['O 0 0 1'])],
                                     testing=True,
                                     queue='short_queue',
@@ -471,5 +479,62 @@ def tearDownClass(cls):
         shutil.rmtree(os.path.join(ARC_TESTING_PATH, 'test_JobAdapter_ServerTimeLimit'), ignore_errors=True)
 
 
+class TestRotateCSV(unittest.TestCase):
+    """
+    Contains unit tests for the CSV rotation logic.
+    """
+
+    def _make_csv(self, path, num_lines):
+        """Helper to create a CSV file with a header and ``num_lines - 1`` data rows."""
+        with open(path, 'w') as f:
+            f.write('col1,col2\n')
+            for i in range(num_lines - 1):
+                f.write(f'{i},data\n')
+
+    def test_no_rotation_below_threshold(self):
+        """Test that no rotation occurs when the file is below the threshold."""
+        with tempfile.TemporaryDirectory() as tmp:
+            csv_path = os.path.join(tmp, 'jobs.csv')
+            self._make_csv(csv_path, 10)
+            JobAdapter._rotate_csv_if_needed(csv_path, max_lines=50)
+            self.assertTrue(os.path.isfile(csv_path))
+            self.assertEqual(glob.glob(os.path.join(tmp, 'jobs.old.*.csv')), [])
+
+    def test_rotation_at_threshold(self):
+        """Test that the file is rotated when it reaches the threshold."""
+        with tempfile.TemporaryDirectory() as tmp:
+            csv_path = os.path.join(tmp, 'jobs.csv')
+            self._make_csv(csv_path, 50)
+            JobAdapter._rotate_csv_if_needed(csv_path, max_lines=50)
+            self.assertFalse(os.path.isfile(csv_path))
+            archives = glob.glob(os.path.join(tmp, 'jobs.old.*.csv'))
+            self.assertEqual(len(archives), 1)
+
+    def test_no_error_for_missing_file(self):
+        """Test that rotation is a no-op when the file does not exist."""
+        JobAdapter._rotate_csv_if_needed('/tmp/nonexistent_arc_test.csv')
+
+    def test_multiple_rotations(self):
+        """Test that multiple rotations produce distinct archive files."""
+        with tempfile.TemporaryDirectory() as tmp:
+            csv_path = os.path.join(tmp, 'jobs.csv')
+            # First rotation on "day 1"
+            self._make_csv(csv_path, 50)
+            with patch('arc.job.adapter.datetime') as mock_dt:
+                mock_dt.datetime.now.return_value = datetime.datetime(2026, 1, 15)
+                mock_dt.timedelta = datetime.timedelta
+                JobAdapter._rotate_csv_if_needed(csv_path, max_lines=50)
+            self.assertFalse(os.path.isfile(csv_path))
+            # Second rotation on "day 2"
+            self._make_csv(csv_path, 50)
+            with patch('arc.job.adapter.datetime') as mock_dt:
+                mock_dt.datetime.now.return_value = datetime.datetime(2026, 2, 20)
+                mock_dt.timedelta = datetime.timedelta
+                JobAdapter._rotate_csv_if_needed(csv_path, max_lines=50)
+            self.assertFalse(os.path.isfile(csv_path))
+            archives = glob.glob(os.path.join(tmp, 'jobs.old.*.csv'))
+            self.assertEqual(len(archives), 2)
+
+
 if __name__ == '__main__':
     unittest.main(testRunner=unittest.TextTestRunner(verbosity=2))