Flow improvements to huey retries

doajtest/unit/test_tasks_ingestarticles_schema_independent.py

@@ -1,14 +1,13 @@
 from doajtest.helpers import DoajTestCase
 from doajtest.mocks.response import ResponseMockFactory
+from portality import models
 
 from portality.tasks import ingestarticles
-
 from portality.bll.services import article as articleSvc
-
-from portality import models
 from portality.core import app
+from portality.background import BackgroundException
 
-from portality.background import BackgroundException, RetryException
+from huey.exceptions import RetryTask
 
 import ftplib, os, requests
 
@@ -29,6 +28,7 @@ def setUp(self):
 
         self.upload_dir = app.config["UPLOAD_DIR"]
         self.ingest_articles_retries = app.config['HUEY_TASKS']['ingest_articles']['retries']
+        self.ingest_articles_retry_delay = app.config['HUEY_TASKS']['ingest_articles']['retry_delay']
 
     def tearDown(self):
         super(TestIngestArticlesSchemaIndependent, self).tearDown()
@@ -41,6 +41,7 @@ def tearDown(self):
 
         app.config["UPLOAD_DIR"] = self.upload_dir
         app.config["HUEY_TASKS"]["ingest_articles"]["retries"] = self.ingest_articles_retries
+        app.config['HUEY_TASKS']['ingest_articles']['retry_delay'] = self.ingest_articles_retry_delay
 
         for id in self.cleanup_ids:
             path = os.path.join(app.config.get("UPLOAD_DIR", "."), id + ".xml")
@@ -111,6 +112,7 @@ def test_2_run_errors(self):
 
     def test_3_submit_retry(self):
         app.config["HUEY_TASKS"]["ingest_articles"]["retries"] = 1
+        app.config['HUEY_TASKS']['ingest_articles']['retry_delay'] = 0
 
         fu = models.FileUpload()
         fu.validated("doaj")
@@ -125,16 +127,9 @@ def test_3_submit_retry(self):
 
         # this assumes that huey is in always eager mode, and thus this immediately calls the async task,
         # which in turn calls execute, which ultimately calls run
-        with self.assertRaises(RetryException):
-            ingestarticles.IngestArticlesBackgroundTask.submit(job)
-
-        job = models.BackgroundJob.pull(job.id)
-        assert job.params.get("ingest_articles__attempts") == 1
-        assert job.status == "processing"
 
-        # now do it again, to see the retry cause the job to fail on the second attempt as per the config
-        with self.assertRaises(RetryException):
-            ingestarticles.IngestArticlesBackgroundTask.submit(job)
+        # Should retry once due to failing to find the file path
+        ingestarticles.IngestArticlesBackgroundTask.submit(job)
 
         job = models.BackgroundJob.pull(job.id)
         assert job.params.get("ingest_articles__attempts") == 2

portality/background.py

@@ -11,6 +11,7 @@
 
 from flask_login import login_user
 from huey import RedisHuey
+from huey.exceptions import RetryTask
 
 from portality import constants
 from portality import models
@@ -24,10 +25,6 @@ class BackgroundException(Exception):
     pass
 
 
-class RetryException(Exception):
-    pass
-
-
 class BackgroundSummary(object):
     def __init__(self, job_id, affected=None, error=None):
         self.job_id = job_id
@@ -70,7 +67,7 @@ def execute(self, background_task: 'BackgroundTask'):
             background_task.run()
             if job.outcome_status == BgjobOutcomeStatus.Pending:
                 job.outcome_status = BgjobOutcomeStatus.Success
-        except RetryException:
+        except RetryTask:
             if job.reference is None:
                 job.reference = {}
             retries = job.reference.get("retries", 0)

portality/tasks/ingestarticles.py

@@ -7,7 +7,8 @@
 import requests
 
 from portality import models
-from portality.background import BackgroundTask, BackgroundApi, BackgroundException, RetryException
+from portality.background import BackgroundTask, BackgroundApi, BackgroundException
+from huey.exceptions import RetryTask
 from portality.bll.exceptions import IngestException
 from portality.core import app
 from portality.lib import plugin
@@ -24,7 +25,7 @@ def load_xwalk(schema):
     try:
         return plugin.load_class(xwalk_name)()
     except IngestException:
-        raise RetryException("Unable to load schema {}".format(xwalk_name))
+        raise RetryTask("Unable to load schema {}".format(xwalk_name))
 
 
 def ftp_upload(job, path, parsed_url, file_upload):
@@ -283,8 +284,8 @@ def _process(self, file_upload: models.FileUpload):
             if retry_limit <= count:
                 job.add_audit_message("File still not found at path {} . Giving up.".format(path))
                 job.fail()
-
-            raise RetryException()
+            else:
+                raise RetryTask()
 
         job.add_audit_message("Importing from {x}".format(x=path))
 
@@ -357,7 +358,7 @@ def submit(cls, background_job):
         :return:
         """
         background_job.save(blocking=True)
-        ingest_articles.schedule(args=(background_job.id,), delay=app.config.get('HUEY_ASYNC_DELAY', 10))
+        ingest_articles.schedule(args=(background_job.id,), delay=app.config.get('HUEY_ASYNC_DELAY', 10), retries=app.config.get("HUEY_TASKS", {}).get("ingest_articles", {}).get("retries", 10), retry_delay=app.config.get("HUEY_TASKS", {}).get("ingest_articles", {}).get("retry_delay", 15))
 
     @classmethod
     def _file_upload(cls, username, f, schema, previous):