Allow hooking job failure for generic error handling

sisyphus/global_settings.py

@@ -60,6 +60,31 @@ def worker_wrapper(job, task_name, call):
     return call
 
 
+def on_job_failure(job):
+    """
+    Job failure hook.
+
+    Can be used for generic job-independent error monitoring, handling or retry
+    logic.
+
+    Sispyhus will call this function w/ the job instance for any failed job.
+
+    The callback itself is then responsible for any retry logic, realized by e.g.
+    analyzing the job log file and removing error files in the job directory as
+    needed.
+
+    The callback needs to be stateless and indempotent, as it can be called multiple
+    times on the same job, especially if the job remains in the error state after the
+    callback has finished.
+
+    Do:
+        - use with caution
+        - ensure you don't build infinite retry loops
+        - limit to specific use cases (e.g. local disk full, GPU broken, etc.)
+    """
+    pass
+
+
 def update_engine_rqmt(last_rqmt: Dict, last_usage: Dict):
     """Update requirements after a job got interrupted, double limits if needed
 

sisyphus/manager.py

@@ -5,6 +5,7 @@
 import sys
 import threading
 import time
+from typing import TYPE_CHECKING, Dict, List
 import warnings
 
 from multiprocessing.pool import ThreadPool
@@ -15,6 +16,9 @@
 from sisyphus.tools import finished_results_cache
 import sisyphus.global_settings as gs
 
+if TYPE_CHECKING:
+    from sisyphus.job import Job
+
 
 class JobCleaner(threading.Thread):
     """Thread to scan all jobs and clean if needed"""
@@ -593,6 +597,7 @@ def run(self):
             self.check_output(write_output=self.link_outputs)
 
             config_manager.continue_readers()
+
             self.update_jobs()
 
             if gs.CLEAR_ERROR or self.clear_errors_once:
@@ -620,6 +625,9 @@ def run(self):
             self.resume_jobs()
             self.run_jobs()
 
+            for job in self.jobs.get(gs.STATE_ERROR, []):
+                gs.on_job_failure(job)
+
         # Stop config reader
         config_manager.cancel_all_reader()