Use async task for ensure_versioned_database_exists

Author: Jakoma02

contentcuration/contentcuration/models.py

@@ -2602,7 +2602,7 @@ class CommunityLibrarySubmission(models.Model):
 
     def save(self, *args, **kwargs):
         # Not a top-level import to avoid circular import issues
-        from contentcuration.utils.publish import ensure_versioned_database_exists
+        from contentcuration.tasks import ensure_versioned_database_exists_task
 
         # Validate on save that the submission author is an editor of the channel
         # and that the version is not greater than the current channel version.
@@ -2630,7 +2630,11 @@ def save(self, *args, **kwargs):
             # When creating a new submission, ensure the channel has a versioned database
             # (it might not have if the channel was published before versioned databases
             # were introduced).
-            ensure_versioned_database_exists(self.channel)
+            ensure_versioned_database_exists_task.fetch_or_enqueue(
+                user=self.author,
+                channel_id=self.channel.id,
+                channel_version=self.channel.version,
+            )
 
         super().save(*args, **kwargs)
 

contentcuration/contentcuration/tasks.py

@@ -19,6 +19,7 @@
 from contentcuration.utils.csv_writer import write_user_csv
 from contentcuration.utils.nodes import calculate_resource_size
 from contentcuration.utils.nodes import generate_diff
+from contentcuration.utils.publish import ensure_versioned_database_exists
 from contentcuration.viewsets.user import AdminUserFilter
 
 
@@ -159,3 +160,8 @@ def sendcustomemails_task(subject, message, query):
             text,
             settings.DEFAULT_FROM_EMAIL,
         )
+
+
+@app.task(name="ensure_versioned_database_exists_task")
+def ensure_versioned_database_exists_task(channel_id, channel_version):
+    ensure_versioned_database_exists(channel_id, channel_version)

contentcuration/contentcuration/tests/helpers.py

@@ -7,6 +7,7 @@
 from django_celery_results.models import TaskResult
 from search.models import ContentNodeFullTextSearch
 
+from contentcuration.celery import app
 from contentcuration.models import ContentNode
 
 
@@ -71,3 +72,27 @@ def reverse_with_query(
     if query:
         return f"{url}?{urlencode(query)}"
     return url
+
+
+class EagerTasksTestMixin(object):
+    """
+    Mixin to make Celery tasks run synchronously during the tests.
+    """
+
+    celery_task_always_eager = None
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        # update celery so tasks are always eager for this test, meaning they'll execute synchronously
+        cls.celery_task_always_eager = app.conf.task_always_eager
+        app.conf.update(task_always_eager=True)
+
+    def setUp(self):
+        super().setUp()
+        clear_tasks()
+
+    @classmethod
+    def tearDownClass(cls):
+        super().tearDownClass()
+        app.conf.update(task_always_eager=cls.celery_task_always_eager)

contentcuration/contentcuration/tests/test_models.py

@@ -34,6 +34,7 @@
 from contentcuration.models import UserHistory
 from contentcuration.tests import testdata
 from contentcuration.tests.base import StudioTestCase
+from contentcuration.tests.helpers import EagerTasksTestMixin
 from contentcuration.viewsets.sync.constants import DELETED
 
 
@@ -550,14 +551,17 @@ def test_make_content_id_unique(self):
 
 
 @mock.patch(
-    "contentcuration.utils.publish.ensure_versioned_database_exists", return_value=None
+    "contentcuration.tasks.ensure_versioned_database_exists_task.fetch_or_enqueue",
+    return_value=None,
 )
-class CommunityLibrarySubmissionTestCase(PermissionQuerysetTestCase):
+class CommunityLibrarySubmissionTestCase(
+    EagerTasksTestMixin, PermissionQuerysetTestCase
+):
     @property
     def base_queryset(self):
         return CommunityLibrarySubmission.objects.all()
 
-    def test_create_submission(self, mock_ensure_db_exists):
+    def test_create_submission(self, mock_ensure_db_exists_task_fetch_or_enqueue):
         # Smoke test
         channel = testdata.channel()
         author = testdata.user()
@@ -587,20 +591,26 @@ def test_save__author_not_editor(self, mock_ensure_db_exists):
         with self.assertRaises(ValidationError):
             submission.save()
 
-    def test_save__nonpositive_channel_version(self, mock_ensure_db_exists):
+    def test_save__nonpositive_channel_version(
+        self, mock_ensure_db_exists_task_fetch_or_enqueue
+    ):
         submission = testdata.community_library_submission()
         submission.channel_version = 0
         with self.assertRaises(ValidationError):
             submission.save()
 
-    def test_save__matching_channel_version(self, mock_ensure_db_exists):
+    def test_save__matching_channel_version(
+        self, mock_ensure_db_exists_task_fetch_or_enqueue
+    ):
         submission = testdata.community_library_submission()
         submission.channel.version = 5
         submission.channel.save()
         submission.channel_version = 5
         submission.save()
 
-    def test_save__impossibly_high_channel_version(self, mock_ensure_db_exists):
+    def test_save__impossibly_high_channel_version(
+        self, mock_ensure_db_exists_task_fetch_or_enqueue
+    ):
         submission = testdata.community_library_submission()
         submission.channel.version = 5
         submission.channel.save()
@@ -609,40 +619,50 @@ def test_save__impossibly_high_channel_version(self, mock_ensure_db_exists):
             submission.save()
 
     def test_save__ensure_versioned_database_exists_on_create(
-        self, mock_ensure_db_exists
+        self, mock_ensure_db_exists_task_fetch_or_enqueue
     ):
         submission = testdata.community_library_submission()
 
-        mock_ensure_db_exists.assert_called_once_with(submission.channel)
+        mock_ensure_db_exists_task_fetch_or_enqueue.assert_called_once_with(
+            user=submission.author,
+            channel_id=submission.channel.id,
+            channel_version=submission.channel.version,
+        )
 
     def test_save__dont_ensure_versioned_database_exists_on_update(
-        self, mock_ensure_db_exists
+        self, mock_ensure_db_exists_task_fetch_or_enqueue
     ):
         submission = testdata.community_library_submission()
-        mock_ensure_db_exists.reset_mock()
+        mock_ensure_db_exists_task_fetch_or_enqueue.reset_mock()
 
         submission.description = "Updated description"
         submission.save()
 
-        mock_ensure_db_exists.assert_not_called()
+        mock_ensure_db_exists_task_fetch_or_enqueue.assert_not_called()
 
-    def test_filter_view_queryset__anonymous(self, mock_ensure_db_exists):
+    def test_filter_view_queryset__anonymous(
+        self, mock_ensure_db_exists_task_fetch_or_enqueue
+    ):
         _ = testdata.community_library_submission()
 
         queryset = CommunityLibrarySubmission.filter_view_queryset(
             self.base_queryset, user=self.anonymous_user
         )
         self.assertFalse(queryset.exists())
 
-    def test_filter_view_queryset__forbidden_user(self, mock_ensure_db_exists):
+    def test_filter_view_queryset__forbidden_user(
+        self, mock_ensure_db_exists_task_fetch_or_enqueue
+    ):
         _ = testdata.community_library_submission()
 
         queryset = CommunityLibrarySubmission.filter_view_queryset(
             self.base_queryset, user=self.forbidden_user
         )
         self.assertFalse(queryset.exists())
 
-    def test_filter_view_queryset__channel_editor(self, mock_ensure_db_exists):
+    def test_filter_view_queryset__channel_editor(
+        self, mock_ensure_db_exists_task_fetch_or_enqueue
+    ):
         submission_a = testdata.community_library_submission()
         submission_b = testdata.community_library_submission()
 
@@ -656,31 +676,39 @@ def test_filter_view_queryset__channel_editor(self, mock_ensure_db_exists):
         self.assertQuerysetContains(queryset, pk=submission_a.id)
         self.assertQuerysetDoesNotContain(queryset, pk=submission_b.id)
 
-    def test_filter_view_queryset__admin(self, mock_ensure_db_exists):
+    def test_filter_view_queryset__admin(
+        self, mock_ensure_db_exists_task_fetch_or_enqueue
+    ):
         submission_a = testdata.community_library_submission()
 
         queryset = CommunityLibrarySubmission.filter_view_queryset(
             self.base_queryset, user=self.admin_user
         )
         self.assertQuerysetContains(queryset, pk=submission_a.id)
 
-    def test_filter_edit_queryset__anonymous(self, mock_ensure_db_exists):
+    def test_filter_edit_queryset__anonymous(
+        self, mock_ensure_db_exists_task_fetch_or_enqueue
+    ):
         _ = testdata.community_library_submission()
 
         queryset = CommunityLibrarySubmission.filter_edit_queryset(
             self.base_queryset, user=self.anonymous_user
         )
         self.assertFalse(queryset.exists())
 
-    def test_filter_edit_queryset__forbidden_user(self, mock_ensure_db_exists):
+    def test_filter_edit_queryset__forbidden_user(
+        self, mock_ensure_db_exists_task_fetch_or_enqueue
+    ):
         _ = testdata.community_library_submission()
 
         queryset = CommunityLibrarySubmission.filter_edit_queryset(
             self.base_queryset, user=self.forbidden_user
         )
         self.assertFalse(queryset.exists())
 
-    def test_filter_edit_queryset__channel_editor(self, mock_ensure_db_exists):
+    def test_filter_edit_queryset__channel_editor(
+        self, mock_ensure_db_exists_task_fetch_or_enqueue
+    ):
         submission = testdata.community_library_submission()
 
         user = testdata.user()
@@ -692,7 +720,9 @@ def test_filter_edit_queryset__channel_editor(self, mock_ensure_db_exists):
         )
         self.assertFalse(queryset.exists())
 
-    def test_filter_edit_queryset__author(self, mock_ensure_db_exists):
+    def test_filter_edit_queryset__author(
+        self, mock_ensure_db_exists_task_fetch_or_enqueue
+    ):
         submission_a = testdata.community_library_submission()
         submission_b = testdata.community_library_submission()
 
@@ -702,15 +732,17 @@ def test_filter_edit_queryset__author(self, mock_ensure_db_exists):
         self.assertQuerysetContains(queryset, pk=submission_a.id)
         self.assertQuerysetDoesNotContain(queryset, pk=submission_b.id)
 
-    def test_filter_edit_queryset__admin(self, mock_ensure_db_exists):
+    def test_filter_edit_queryset__admin(
+        self, mock_ensure_db_exists_task_fetch_or_enqueue
+    ):
         submission_a = testdata.community_library_submission()
 
         queryset = CommunityLibrarySubmission.filter_edit_queryset(
             self.base_queryset, user=self.admin_user
         )
         self.assertQuerysetContains(queryset, pk=submission_a.id)
 
-    def test_mark_live(self, mock_ensure_db_exists):
+    def test_mark_live(self, mock_ensure_db_exists_task_fetch_or_enqueue):
         submission_a = testdata.community_library_submission()
         submission_b = testdata.community_library_submission()
 

contentcuration/contentcuration/tests/viewsets/base.py

@@ -2,9 +2,8 @@
 
 from django.urls import reverse
 
-from contentcuration.celery import app
 from contentcuration.models import Change
-from contentcuration.tests.helpers import clear_tasks
+from contentcuration.tests.helpers import EagerTasksTestMixin
 from contentcuration.viewsets.sync.constants import CHANNEL
 from contentcuration.viewsets.sync.constants import SYNCED
 from contentcuration.viewsets.sync.utils import _generate_event as base_generate_event
@@ -102,25 +101,7 @@ def generate_publish_next_event(channel_id, use_staging_tree=False):
     return event
 
 
-class SyncTestMixin(object):
-    celery_task_always_eager = None
-
-    @classmethod
-    def setUpClass(cls):
-        super(SyncTestMixin, cls).setUpClass()
-        # update celery so tasks are always eager for this test, meaning they'll execute synchronously
-        cls.celery_task_always_eager = app.conf.task_always_eager
-        app.conf.update(task_always_eager=True)
-
-    def setUp(self):
-        super(SyncTestMixin, self).setUp()
-        clear_tasks()
-
-    @classmethod
-    def tearDownClass(cls):
-        super(SyncTestMixin, cls).tearDownClass()
-        app.conf.update(task_always_eager=cls.celery_task_always_eager)
-
+class SyncTestMixin(EagerTasksTestMixin):
     @property
     def sync_url(self):
         return reverse("sync")

contentcuration/contentcuration/utils/publish.py

@@ -1094,32 +1094,39 @@ def publish_channel(  # noqa: C901
     return channel
 
 
-def ensure_versioned_database_exists(channel):
+def ensure_versioned_database_exists(channel_id, channel_version):
     """
     Ensures that the versioned database exists, and if not, copies the unversioned database to the versioned path.
     This happens if the channel was published back when versioned databases were not used.
     """
-    if channel.version == 0:
+    if channel_version == 0:
         raise ValueError("An unpublished channel cannot have a versioned database.")
 
     unversioned_db_storage_path = os.path.join(
-        settings.DB_ROOT, "{id}.sqlite3".format(id=channel.id)
+        settings.DB_ROOT, "{id}.sqlite3".format(id=channel_id)
     )
     versioned_db_storage_path = os.path.join(
         settings.DB_ROOT,
-        "{id}-{version}.sqlite3".format(id=channel.id, version=channel.version),
+        "{id}-{version}.sqlite3".format(id=channel_id, version=channel_version),
     )
 
     if not storage.exists(versioned_db_storage_path):
         if not storage.exists(unversioned_db_storage_path):
             # This should never happen, a published channel should always have an unversioned database
             raise FileNotFoundError(
-                f"Neither unversioned nor versioned database found for channel {channel.id}."
+                f"Neither unversioned nor versioned database found for channel {channel_id}."
             )
 
+        # NOTE: This should not result in a race condition in the case that a newer
+        # version of the channel is published before the task running this function
+        # is executed. In that case, the publishing logic would have already created
+        # the versioned database. The only case where this could be problematic is
+        # if this happens between the check above this comment and the commands below
+        # it. However, this is EXTREMELY unlikely, and could probably only be solved
+        # by introducing a locking mechanism for the database storage objects.
         with storage.open(unversioned_db_storage_path, "rb") as unversioned_db_file:
             storage.save(versioned_db_storage_path, unversioned_db_file)
 
         logging.info(
-            f"Versioned database for channel {channel.id} did not exist, copied the unversioned database to {versioned_db_storage_path}."
+            f"Versioned database for channel {channel_id} did not exist, copied the unversioned database to {versioned_db_storage_path}."
         )