Handled removed target from workflow executor #1189

src/sos/step_executor.py

@@ -1296,6 +1296,11 @@ def run(self):
                         if e.message:
                             env.logger.info(e)
                         self.output_groups[proc_result['index']] = sos_targets([])
+                    #elif isinstance(e, RemovedTarget):
+                    #
+                    # in theory, we should be able to handled removed target from here
+                    # by rerunning the substep, but we it is too much work for this
+                    # corner case. Let us simply rerunt he entire step.
                     else:
                         raise e
             # if output is Undetermined, re-evalulate it

src/sos/workflow_executor.py

@@ -1165,6 +1165,8 @@ def i_am():
                         runnable._child_socket.close()
                     elif isinstance(res, UnavailableLock):
                         self.handle_unavailable_lock(res, dag, runnable)
+                    elif isinstance(res, RemovedTarget):
+                        self.handle_unknown_target(res.target, dag, runnable)
                     # if the job is failed
                     elif isinstance(res, Exception):
                         env.logger.debug(f'{i_am()} received an exception')
@@ -1422,6 +1424,13 @@ def i_am():
                         f'{i_am()} receive a result {short_repr(res)}')
                     if isinstance(res, UnavailableLock):
                         self.handle_unavailable_lock(res, dag, runnable)
+                    elif isinstance(res, RemovedTarget):
+                        # RemovedTarget is usually hanled at the step level
+                        # by sending a missing-target message here. However,
+                        # if the exception is raised from substep workers, it is
+                        # difficult for a substep to rerun particular substep
+                        # so we handle it here by rerunning the entire step
+                        self.handle_unknown_target(res.target, dag, runnable)
                     # if the job is failed
                     elif isinstance(res, Exception):
                         env.logger.debug(f'{i_am()} received an exception')