Added checks for existing long-running tasks on publish

contentcuration/contentcuration/tests/test_exportchannel.py

@@ -24,6 +24,7 @@
 from contentcuration.utils.publish import MIN_SCHEMA_VERSION
 from contentcuration.utils.publish import prepare_export_database
 from contentcuration.utils.publish import set_channel_icon_encoding
+from contentcuration.utils.publish import wait_for_async_tasks
 
 pytestmark = pytest.mark.django_db
 
@@ -192,6 +193,38 @@ def test_create_slideshow_manifest(self):
         manifest_collection = cc.File.objects.filter(contentnode=ccnode, preset_id=u"slideshow_manifest")
         assert len(manifest_collection) == 1
 
+    def test_blocking_task_detection(self):
+        with patch('time.sleep') as patched_time_sleep:
+            user = cc.User.objects.create()
+            channel = cc.Channel.objects.create()
+            cc.Task.objects.create(channel_id=channel.pk, user_id=user.pk, task_type='sync-channel', metadata={})
+            wait_for_async_tasks(channel, attempts=1)
+            self.assertEqual(1, patched_time_sleep.call_count)
+
+    def test_blocking_task_completion_detection(self):
+        with patch('time.sleep') as patched_time_sleep:
+            user = cc.User.objects.create()
+            channel = cc.Channel.objects.create()
+            cc.Task.objects.create(channel_id=channel.pk, user_id=user.pk, task_type='sync-channel', metadata={}, status='SUCCESS')
+            wait_for_async_tasks(channel, attempts=1)
+            self.assertEqual(0, patched_time_sleep.call_count)
+
+    def test_blocking_task_failure_detection(self):
+        with patch('time.sleep') as patched_time_sleep:
+            user = cc.User.objects.create()
+            channel = cc.Channel.objects.create()
+            cc.Task.objects.create(channel_id=channel.pk, user_id=user.pk, task_type='sync-channel', metadata={}, status='FAILURE')
+            wait_for_async_tasks(channel, attempts=1)
+            self.assertEqual(0, patched_time_sleep.call_count)
+
+    def test_nonblocking_task_detection(self):
+        with patch('time.sleep') as patched_time_sleep:
+            user = cc.User.objects.create()
+            channel = cc.Channel.objects.create()
+            cc.Task.objects.create(channel_id=channel.pk, user_id=user.pk, task_type='get-node-diff', metadata={})
+            wait_for_async_tasks(channel, attempts=1)
+            self.assertEqual(0, patched_time_sleep.call_count)
+
 
 class ChannelExportPrerequisiteTestCase(StudioTestCase):
     @classmethod

contentcuration/contentcuration/utils/publish.py

@@ -8,6 +8,7 @@
 import os
 import re
 import tempfile
+import time
 import traceback
 import uuid
 import zipfile
@@ -51,6 +52,7 @@
 PERSEUS_IMG_DIR = exercises.IMG_PLACEHOLDER + "/images"
 THUMBNAIL_DIMENSION = 128
 MIN_SCHEMA_VERSION = "1"
+BLOCKING_TASK_TYPES = ["duplicate-nodes", "move-nodes", "sync-channel"]
 
 
 def send_emails(channel, user_id, version_notes=''):
@@ -689,12 +691,30 @@ def fill_published_fields(channel, version_notes):
     channel.save()
 
 
+def _check_for_blocking_tasks(channel):
+    return ccmodels.Task.objects.filter(
+        task_type__in=BLOCKING_TASK_TYPES,
+        channel_id=channel.pk,
+    ).exclude(status='FAILURE').exclude(status='SUCCESS').exists()
+
+
+# Default try for 30 minutes
+def wait_for_async_tasks(channel, attempts=360):
+    while attempts and _check_for_blocking_tasks(channel):
+        attempts -= 1
+        time.sleep(5)
+
+    if not attempts and _check_for_blocking_tasks(channel):
+        logging.warning('Ran out of attempts: Tasks still detected for {} during publish'.format(channel.pk))
+
+
 def publish_channel(user_id, channel_id, version_notes='', force=False, force_exercises=False, send_email=False, task_object=None):
     channel = ccmodels.Channel.objects.get(pk=channel_id)
     kolibri_temp_db = None
 
     try:
         set_channel_icon_encoding(channel)
+        wait_for_async_tasks(channel)
         kolibri_temp_db = create_content_database(channel, force, user_id, force_exercises, task_object)
         increment_channel_version(channel)
         mark_all_nodes_as_published(channel)