Bug: Extend error handling in backend_write_file() (#983)

* Bug: Extend error handling in backend_write_file()

* dump first

Author: jan-janssen

src/executorlib/task_scheduler/file/backend.py

@@ -45,15 +45,22 @@ def backend_write_file(file_name: str, output: Any, runtime: float) -> None:
     """
     file_name_out = os.path.splitext(file_name)[0][:-2]
     os.rename(file_name, file_name_out + "_r.h5")
-    if "result" in output:
-        dump(
-            file_name=file_name_out + "_r.h5",
-            data_dict={"output": output["result"], "runtime": runtime},
-        )
-    else:
+    try:
+        if "result" in output:
+            dump(
+                file_name=file_name_out + "_r.h5",
+                data_dict={"output": output["result"], "runtime": runtime},
+            )
+        else:
+            dump(
+                file_name=file_name_out + "_r.h5",
+                data_dict={"error": output["error"], "runtime": runtime},
+            )
+    except Exception as serialize_error:
+        # Serialization failed — store the error so the job is not stuck
         dump(
             file_name=file_name_out + "_r.h5",
-            data_dict={"error": output["error"], "runtime": runtime},
+            data_dict={"error": serialize_error, "runtime": runtime},
         )
     os.rename(file_name_out + "_r.h5", file_name_out + "_o.h5")
 

tests/unit/task_scheduler/file/test_backend.py

@@ -7,7 +7,7 @@
 
 
 try:
-    from executorlib.task_scheduler.file.backend import backend_execute_task_in_file
+    from executorlib.task_scheduler.file.backend import backend_execute_task_in_file, backend_write_file
     from executorlib.task_scheduler.file.shared import _check_task_output, _convert_args_and_kwargs, FutureItem
     from executorlib.standalone.hdf import dump, get_runtime
     from executorlib.standalone.serialize import serialize_funct
@@ -64,6 +64,30 @@ def test_execute_function_mixed(self):
         self.assertTrue(future_file_obj.done())
         self.assertEqual(future_file_obj.result(), 3)
 
+    def test_backend_write_file(self):
+        cache_directory = os.path.abspath("executorlib_cache")
+        os.makedirs(cache_directory, exist_ok=True)
+        file_name = os.path.join(cache_directory, "test_file_i.h5")
+        dump(file_name=file_name, data_dict={"fn": my_funct, "args": [1], "kwargs": {"b": 2}})
+        backend_write_file(file_name=file_name, output={"result": 3}, runtime=0.1)
+        future_file_obj = FutureItem(
+            file_name=os.path.join(cache_directory, "test_file_o.h5")
+        )
+        self.assertTrue(future_file_obj.done())
+        self.assertEqual(future_file_obj.result(), 3)
+
+    def test_backend_write_file_serialization_error(self):
+        cache_directory = os.path.abspath("executorlib_cache")
+        os.makedirs(cache_directory, exist_ok=True)
+        file_name = os.path.join(cache_directory, "test_file_i.h5")
+        dump(file_name=file_name, data_dict={"fn": my_funct, "args": [1], "kwargs": {"b": 2}})
+        backend_write_file(file_name=file_name, output={"result": Future()}, runtime=0.1)
+        future_file_obj = FutureItem(
+            file_name=os.path.join(cache_directory, "test_file_o.h5")
+        )
+        with self.assertRaises(Exception):
+            future_file_obj.result()
+
     def test_execute_function_mixed_selector_convert(self):
         cache_directory = os.path.abspath("executorlib_cache")
         os.makedirs(cache_directory, exist_ok=True)