Fix revoking tasks on custom queues

src/apps/competitions/models.py

@@ -12,7 +12,7 @@
 from django.urls import reverse
 from django.utils.timezone import now
 
-from celery_config import app
+from celery_config import app, app_for_vhost
 from chahub.models import ChaHubSaveMixin
 from leaderboards.models import SubmissionScore
 from profiles.models import User, Organization
@@ -644,7 +644,12 @@ def cancel(self, status=CANCELLED):
             if self.has_children:
                 for sub in self.children.all():
                     sub.cancel(status=status)
-            app.control.revoke(self.celery_task_id, terminate=True)
+            celery_app = app
+            # If a custom queue is set, we need to fetch the appropriate celery app
+            if self.phase.competition.queue:
+                celery_app = app_for_vhost(str(self.phase.competition.queue.vhost))
+
+            celery_app.control.revoke(self.celery_task_id, terminate=True)
             self.status = status
             self.save()
             return True

src/celery_config.py

@@ -1,5 +1,8 @@
 from celery import Celery
 from kombu import Queue, Exchange
+from django.conf import settings
+import urllib.parse
+import copy
 
 app = Celery()
 
@@ -11,3 +14,25 @@
     # Mostly defining queue here so we can set x-max-priority
     Queue('compute-worker', Exchange('compute-worker'), routing_key='compute-worker', queue_arguments={'x-max-priority': 10}),
 ]
+
+_vhost_apps = {}
+# Function to get the app for a vhost
+def app_for_vhost(vhost):
+    if vhost not in _vhost_apps:
+        # Take the CELERY_BROKER_URL and replace the vhost with the vhhost for this queue
+        broker_url = settings.CELERY_BROKER_URL
+        # This is require to work around https://bugs.python.org/issue18828
+        scheme = urllib.parse.urlparse(broker_url).scheme
+        urllib.parse.uses_relative.append(scheme)
+        urllib.parse.uses_netloc.append(scheme)
+        broker_url = urllib.parse.urljoin(broker_url, vhost)
+
+        vhost_app = Celery()
+        # Copy the settings so we can modify the broker url to include the vhost
+        django_settings = copy.copy(settings)
+        django_settings.CELERY_BROKER_URL = broker_url
+        vhost_app.config_from_object(django_settings, namespace='CELERY')
+        vhost_app.task_queues = app.conf.task_queues
+        _vhost_apps[vhost] = vhost_app
+
+    return _vhost_apps[vhost]