Add retry on activity worker's fail

Totally untested!

Signed-off-by: Yves Bastide <yves@botify.com>

Author: ybastide

simpleflow/swf/process/poller.py

@@ -154,3 +154,18 @@ def poll_with_retry(self):
         )(self.poll)
         response = poll(task_list, identity=identity)
         return response
+
+    @abc.abstractmethod
+    def fail(self, *args, **kwargs):
+        """fail; only relevant for activity workers."""
+        raise NotImplementedError
+
+    def fail_with_retry(self, *args, **kwargs):
+        fail = utils.retry.with_delay(
+            nb_times=self.nb_retries,
+            delay=utils.retry.exponential,
+            log_with=logger.exception,
+            on_exceptions=swf.exceptions.ResponseError,
+        )(self.fail)
+        response = fail(*args, **kwargs)
+        return response

simpleflow/swf/process/worker/base.py

@@ -142,7 +142,7 @@ def process(self, poller, token, task):
         except Exception as err:
             logger.exception("process error: {}".format(str(err)))
             tb = traceback.format_exc()
-            return poller.fail(token, task, reason=str(err), details=tb)
+            return poller.fail_with_retry(token, task, reason=str(err), details=tb)
 
         try:
             poller.complete_with_retry(token, json_dumps(result))
@@ -152,7 +152,7 @@ def process(self, poller, token, task):
                 task.activity_id,
                 err,
             )
-            poller.fail(token, task, reason)
+            poller.fail_with_retry(token, task, reason)
 
 
 def process_task(poller, token, task):
@@ -204,7 +204,7 @@ def worker_alive():
                         worker.pid
                     ))
             if worker.exitcode != 0:
-                poller.fail(
+                poller.fail_with_retry(
                     token,
                     task,
                     reason='process {} died: exit code {}'.format(