diff --git a/src/LibreQoS.py b/src/LibreQoS.py index f5244a6a..a0b9e08c 100755 --- a/src/LibreQoS.py +++ b/src/LibreQoS.py @@ -227,6 +227,20 @@ def ensure_parent_dir(path): os.makedirs(parent, exist_ok=True) +def atomic_write_text(path, text): + # Write to a sibling temp file and atomically replace the target, so an + # interrupted write (crash, VM snapshot) can never leave a truncated file. + # The file is fsync'd, but the directory rename is not; on power loss the + # worst case is the previous complete file. + ensure_parent_dir(path) + temp_path = path + ".tmp" + with open(temp_path, 'w') as file: + file.write(text) + file.flush() + os.fsync(file.fileno()) + os.replace(temp_path, path) + + def get_network_json_path(): base_dir = get_libreqos_directory() effective_path = get_runtime_state_path("topology", "network.effective.json") @@ -641,13 +655,7 @@ def save_planner_state(state, state_path=None, planner_module=None): planner_module.save_state(state_path, state) return - parent = os.path.dirname(state_path) - if parent: - os.makedirs(parent, exist_ok=True) - temp_path = state_path + ".tmp" - with open(temp_path, "w") as outfile: - json.dump(state, outfile, indent=2, sort_keys=True) - os.replace(temp_path, state_path) + atomic_write_text(state_path, json.dumps(state, indent=2, sort_keys=True)) def sanitize_planner_state(state): @@ -927,19 +935,22 @@ def shellReturn(command): def checkIfFirstRunSinceBoot(): last_run_path = get_last_run_path() + lastRun = None if os.path.isfile(last_run_path): - with open(last_run_path, 'r') as file: - lastRun = datetime.strptime(file.read(), "%d-%b-%Y (%H:%M:%S.%f)") + try: + with open(last_run_path, 'r') as file: + lastRun = datetime.strptime(file.read(), "%d-%b-%Y (%H:%M:%S.%f)") + except (ValueError, OSError) as err: + # Torn/empty/corrupt lastRun.txt (e.g. interrupted write). Treat as + # first-run-since-boot rather than killing the whole scheduler run. + logging.warning("lastRun.txt at '%s' is unreadable (%s); assuming first run since boot.", last_run_path, err) + if lastRun is not None: systemRunningSince = datetime.fromtimestamp(psutil.boot_time()) - if systemRunningSince > lastRun: - print("First time run since system boot.") - return True - else: + if systemRunningSince <= lastRun: print("Not first time run since system boot.") return False - else: - print("First time run since system boot.") - return True + print("First time run since system boot.") + return True def clearPriorSettings(interfaceA, interfaceB): if enable_actual_shell_commands(): @@ -2872,10 +2883,8 @@ def refreshShapers(): queuingStructure['generatedPNs'] = generatedPNs queuingStructure['logical_to_physical_node'] = logical_to_physical_node queuingStructure['virtual_nodes'] = virtual_nodes - queuing_structure_path = get_state_path("shaping", "queuingStructure.json") - ensure_parent_dir(queuing_structure_path) - with open(queuing_structure_path, 'w') as infile: - json.dump(queuingStructure, infile, indent=4) + queuing_structure_path = get_queuing_structure_path() + atomic_write_text(queuing_structure_path, json.dumps(queuingStructure, indent=4)) # Record start time of actual filter reload @@ -3009,22 +3018,15 @@ def refreshShapers(): shutil.copyfile(shapedDevicesFile, last_loaded_path) # Save for stats - stats_by_circuit_path = get_state_path("stats", "statsByCircuit.json") - ensure_parent_dir(stats_by_circuit_path) - with open(stats_by_circuit_path, 'w') as f: - f.write(json.dumps(subscriberCircuits, indent=4)) - stats_by_parent_node_path = get_state_path("stats", "statsByParentNode.json") - ensure_parent_dir(stats_by_parent_node_path) - with open(stats_by_parent_node_path, 'w') as f: - f.write(json.dumps(parentNodes, indent=4)) + stats_by_circuit_path = get_stats_by_circuit_path() + atomic_write_text(stats_by_circuit_path, json.dumps(subscriberCircuits, indent=4)) + stats_by_parent_node_path = get_stats_by_parent_node_path() + atomic_write_text(stats_by_parent_node_path, json.dumps(parentNodes, indent=4)) # Record time this run completed at - # filename = os.path.join(_here, 'lastRun.txt') - last_run_path = get_state_path("stats", "lastRun.txt") - ensure_parent_dir(last_run_path) - with open(last_run_path, 'w') as file: - file.write(datetime.now().strftime("%d-%b-%Y (%H:%M:%S.%f)")) + last_run_path = get_last_run_path() + atomic_write_text(last_run_path, datetime.now().strftime("%d-%b-%Y (%H:%M:%S.%f)")) # Report reload time diff --git a/src/test_libreqos_shaping_inputs.py b/src/test_libreqos_shaping_inputs.py index 2bc8d99d..83f681b6 100644 --- a/src/test_libreqos_shaping_inputs.py +++ b/src/test_libreqos_shaping_inputs.py @@ -1,4 +1,5 @@ import csv +import datetime import importlib import json import os @@ -521,5 +522,85 @@ class TestLibreQoSShapingInputs(unittest.TestCase): self.assertEqual(missing_parents, {}) +class TestFirstRunSinceBoot(unittest.TestCase): + def setUp(self): + temp_dir = tempfile.TemporaryDirectory() # nosec B108 + self.addCleanup(temp_dir.cleanup) + self.last_run_path = os.path.join(temp_dir.name, "lastRun.txt") + patcher = patch.object( + LibreQoS, "get_last_run_path", return_value=self.last_run_path + ) + patcher.start() + self.addCleanup(patcher.stop) + + def test_missing_file_is_first_run(self): + self.assertTrue(LibreQoS.checkIfFirstRunSinceBoot()) + + def test_empty_file_is_first_run(self): + # Regression: a torn write (e.g. interrupted VM snapshot) used to crash + # the scheduler with ValueError from strptime. + with open(self.last_run_path, "w", encoding="utf-8"): + pass + with self.assertLogs(level="WARNING"): + self.assertTrue(LibreQoS.checkIfFirstRunSinceBoot()) + + def test_garbage_file_is_first_run(self): + with open(self.last_run_path, "w", encoding="utf-8") as handle: + handle.write("\x00partial wr") + self.assertTrue(LibreQoS.checkIfFirstRunSinceBoot()) + + def test_unreadable_file_is_first_run(self): + with open(self.last_run_path, "w", encoding="utf-8"): + pass + with patch("builtins.open", side_effect=PermissionError(13, "denied")): + self.assertTrue(LibreQoS.checkIfFirstRunSinceBoot()) + + def test_recent_run_is_not_first_run(self): + with open(self.last_run_path, "w", encoding="utf-8") as handle: + handle.write( + datetime.datetime.now().strftime("%d-%b-%Y (%H:%M:%S.%f)") + ) + self.assertFalse(LibreQoS.checkIfFirstRunSinceBoot()) + + def test_run_before_boot_is_first_run(self): + old = datetime.datetime.now() - datetime.timedelta(days=3650) + with open(self.last_run_path, "w", encoding="utf-8") as handle: + handle.write(old.strftime("%d-%b-%Y (%H:%M:%S.%f)")) + self.assertTrue(LibreQoS.checkIfFirstRunSinceBoot()) + + +class TestAtomicWriteText(unittest.TestCase): + def setUp(self): + temp_dir = tempfile.TemporaryDirectory() # nosec B108 + self.addCleanup(temp_dir.cleanup) + self.temp_dir = temp_dir.name + + def test_atomic_write_creates_parents_and_content(self): + path = os.path.join(self.temp_dir, "a", "b", "state.txt") + LibreQoS.atomic_write_text(path, "hello") + with open(path, encoding="utf-8") as handle: + self.assertEqual(handle.read(), "hello") + self.assertFalse(os.path.exists(path + ".tmp")) + + def test_atomic_write_replaces_existing_content(self): + path = os.path.join(self.temp_dir, "state.txt") + with open(path, "w", encoding="utf-8") as handle: + handle.write("old") + LibreQoS.atomic_write_text(path, "new") + with open(path, encoding="utf-8") as handle: + self.assertEqual(handle.read(), "new") + self.assertFalse(os.path.exists(path + ".tmp")) + + def test_atomic_write_failure_preserves_existing_content(self): + path = os.path.join(self.temp_dir, "state.txt") + with open(path, "w", encoding="utf-8") as handle: + handle.write("old") + with patch("os.replace", side_effect=OSError(5, "replace failed")): + with self.assertRaises(OSError): + LibreQoS.atomic_write_text(path, "new") + with open(path, encoding="utf-8") as handle: + self.assertEqual(handle.read(), "old") + + if __name__ == "__main__": unittest.main()