Correctly retry failed batch index updates

Using retry doesn't work in celery-batches, it just raises the original exception. See clokep/celery-batches#10 Fixes WEBLATE-FM Fixes WEBLATE-FY Fixes WEBLATE-FW Fixes WEBLATE-FT Fixes WEBLATE-FS Fixes WEBLATE-FV Fixes WEBLATE-FR Signed-off-by: Michal Čihař <[email protected]>
WeblateOrg · Oct 9, 2018 · f112364 · f112364
1 parent 640e327
commit f112364
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 18 deletions.
diff --git a/docs/changes.rst b/docs/changes.rst
@@ -12,6 +12,7 @@ Released on ? 2018.
 * Fixed progress reporting for newly added languages.
 * Correctly report Celery worker errors to Sentry.
 * Fixed creating new translations with Qt Linguist.
+* Fixed occasional fulltext index update failures.
 
 weblate 3.2
 ------------

diff --git a/weblate/memory/tasks.py b/weblate/memory/tasks.py
@@ -80,10 +80,7 @@ def update_memory(user, unit):
         )
 
 
-@app.task(
-    base=Batches, flush_every=1000, flush_interval=300, bind=True,
-    max_retries=1000
-)
+@app.task(base=Batches, flush_every=1000, flush_interval=300, bind=True)
 def update_memory_task(self, *args, **kwargs):
     def fixup_strings(data):
         result = {}
@@ -101,8 +98,12 @@ def fixup_strings(data):
         with memory.writer() as writer:
             for item in data:
                 writer.add_document(**fixup_strings(item))
-    except LockError as exc:
-        raise self.retry(exc=exc)
+    except LockError:
+        # Manually handle retries, it doesn't work
+        # with celery-batches
+        sleep(10)
+        for unit in data:
+            update_memory_task.delay(**unit)
 
 
 @app.task

diff --git a/weblate/trans/search.py b/weblate/trans/search.py
@@ -23,6 +23,7 @@
 from __future__ import absolute_import, unicode_literals
 
 import functools
+from time import sleep
 
 from celery_batches import Batches
 
@@ -250,25 +251,23 @@ def delete_search_units(self, source_units, languages):
                     writer.delete_by_term('pk', pk)
 
 
-@app.task(
-    base=Batches, flush_every=1000, flush_interval=300, bind=True,
-    max_retries=1000
-)
+@app.task(base=Batches, flush_every=1000, flush_interval=300, bind=True)
 def update_fulltext(self, *args, **kwargs):
     unitdata = extract_batch_kwargs(*args, **kwargs)
     fulltext = Fulltext()
 
     # Update index
     try:
         fulltext.update_index(unitdata)
-    except LockError as exc:
-        raise self.retry(exc=exc)
+    except LockError:
+        # Manually handle retries, it doesn't work
+        # with celery-batches
+        sleep(10)
+        for unit in unitdata:
+            update_fulltext.delay(**unit)
 
 
-@app.task(
-    base=Batches, flush_every=1000, flush_interval=300, bind=True,
-    max_retries=1000
-)
+@app.task(base=Batches, flush_every=1000, flush_interval=300, bind=True)
 def delete_fulltext(self, *args):
     ids = extract_batch_args(*args)
     fulltext = Fulltext()
@@ -285,5 +284,9 @@ def delete_fulltext(self, *args):
 
     try:
         fulltext.delete_search_units(units, languages)
-    except LockError as exc:
-        raise self.retry(exc=exc)
+    except LockError:
+        # Manually handle retries, it doesn't work
+        # with celery-batches
+        sleep(10)
+        for unit in ids:
+            delete_fulltext.delay(*unit)