Merge pull request #284 from buildingSMART/IVS-720_Fix_Platform_Fairness_via_Locks

Author: rw-bsi

backend/apps/ifc_validation/tasks/task_runner.py

@@ -1,10 +1,11 @@
 import functools
 import psutil
+import random
 
 from celery import shared_task, chain, chord, group
 from celery.exceptions import SoftTimeLimitExceeded
 
-from core.redis_lock import acquire_user_lock
+from core.redis_lock import acquire_user_lock, LockError
 from core.utils import log_execution
 
 from apps.ifc_validation_models.decorators import requires_django_user_context
@@ -66,10 +67,14 @@ def wrapper(self, *args, **kwargs):
                 with acquire_user_lock(user_id, task_name):
                     return task_func(self, *args, **kwargs)
 
-            except RuntimeError as exc:
-                # countdown: exponential backoff - prevents thundering herd
-                # max_retries=0: infinite retries - only for locking purposes
-                raise self.retry(exc=exc, countdown=15 + self.request.retries * 7, max_retries=0)
+            except LockError as exc:
+                base_backoff = 10 + self.request.retries * 7
+                jitter_backoff = random.randint(0,10)
+                raise self.retry(
+                    exc=exc, 
+                    countdown=base_backoff+jitter_backoff, 
+                    max_retries=None # infinite
+                )
 
             except Exception as exc:
                 # other errors — fail permanently
@@ -163,11 +168,11 @@ def on_workflow_failed(self, *args, **kwargs):
 def task_factory(task_type):
     config = task_registry[task_type]
 
-    @shared_task(bind=True, name=config.celery_task_name)
+    @shared_task(bind=True, name=config.celery_task_name, max_retries=None)
+    @with_user_task_lock(task_name=config.celery_task_name)
     @log_execution
     @requires_django_user_context
     @kill_subprocesses_on_timeout
-    @with_user_task_lock(task_name=config.celery_task_name)
     def validation_subtask_runner(self, *args, **kwargs):
 
         id = kwargs.get('id')

backend/core/celery.py

@@ -28,7 +28,12 @@ def on_failure(self, exc_info, send_failed_event=True, return_ok=False):
             send_failed_event=send_failed_event,
             return_ok=return_ok
         )
-        logger.error(f'Failure detected for task {self.task.name} - {exc_info=}')
+        logger.error(f'Failure detected for task {self.task.name} - {exc_info}')
+
+    def on_retry(self, exc_info):
+        
+        super().on_retry(exc_info)
+        logger.warning(f"Retry detected for task {self.task.name} - {exc_info}")
 
 
 class BaseTask(Task):
@@ -49,7 +54,7 @@ def on_success(self, retval, task_id, args, kwargs):
         logger.debug("*** SUCCESS ***")
 
     def on_retry(self, exc, task_id, args, kwargs, einfo):
-        logger.warn("*** RETRY ***")
+        logger.warning("*** RETRY ***")
 
     def on_failure(self, exc, task_id, args, kwargs, einfo):
         logger.error("*** FAILURE ***")

backend/core/redis_lock.py

@@ -9,6 +9,10 @@
 redis_client: Redis = Redis.from_url(CELERY_BROKER_URL, decode_responses=True)
 
 
+class LockError(Exception):
+    pass
+
+
 @contextmanager
 def acquire_user_lock(
     user_id: int | str,
@@ -32,7 +36,7 @@ def acquire_user_lock(
 
     acquired = lock.acquire(blocking=True)
     if not acquired:
-        raise RuntimeError(f"User {user_id} already has an active task_name {task_name} (lock held).")
+        raise LockError(f"User {user_id} already has an active task_name {task_name} (lock held).")
 
     try:
         logger.info(f"Lock acquired for user {user_id} and task {task_name} (key={lock_key})")

backend/core/settings.py

@@ -338,6 +338,7 @@
 CELERY_TASK_ALLOW_ERROR_CB_ON_CHORD_HEADER = True
 
 # reliability settings - see https://www.francoisvoron.com/blog/configure-celery-for-reliable-delivery
+CELERY_WORKER_PREFETCH_MULTIPLIER = 1
 CELERY_TASK_REJECT_ON_WORKER_LOST = True
 CELERY_TASK_ACKS_LATE = True
 CELERY_TASK_STORE_ERRORS_EVEN_IF_IGNORED = True