fix: pin bulk submission delete to primary DB

onadata/apps/api/tasks.py

@@ -17,6 +17,7 @@
 from django.utils.datastructures import MultiValueDict
 
 from celery.result import AsyncResult
+from multidb.pinning import use_master
 
 from onadata.apps.api import tools
 from onadata.apps.logger.models import Instance, Project, ProjectInvitation, XForm
@@ -213,6 +214,7 @@ def share_project_async(project_id, username, role, remove=False):
 
 
 @app.task(base=AutoRetryTask)
+@use_master
 def delete_xform_submissions_async(
     xform_id: int,
     deleted_by_id: int,

onadata/apps/api/tests/test_tasks.py

@@ -7,6 +7,8 @@
 from django.core.cache import cache
 from django.db import DatabaseError, OperationalError
 
+from multidb.pinning import this_thread_is_pinned
+
 from onadata.apps.api.tasks import (
     ShareProject,
     delete_xform_submissions_async,
@@ -231,3 +233,11 @@ def test_user_id_invalid(self, mock_logger, mock_delete):
         delete_xform_submissions_async.delay(self.xform.pk, sys.maxsize)
         self.assertFalse(mock_delete.called)
         mock_logger.assert_called_once()
+
+    def test_pinned_to_primary_db(self, mock_delete):
+        """Thread is pinned to primary DB during task execution"""
+        mock_delete.side_effect = lambda *a, **kw: self.assertTrue(
+            this_thread_is_pinned(),
+            "Thread was not pinned to primary DB when calling delete_xform_submissions",
+        )
+        delete_xform_submissions_async.delay(self.xform.pk, self.user.pk)

onadata/libs/tests/utils/test_logger_tools.py

@@ -19,6 +19,7 @@
 
 from azure.storage.blob import AccountSasPermissions
 from defusedxml.ElementTree import ParseError
+from multidb.pinning import this_thread_is_pinned
 
 from onadata.apps.logger.import_tools import django_file
 from onadata.apps.logger.models import Instance, InstanceHistory
@@ -1186,6 +1187,27 @@ def test_decrypted_submission_count_updated(self):
         self.xform.refresh_from_db()
         self.assertEqual(self.xform.num_of_decrypted_submissions, 3)
 
+    def test_pinned_to_primary_db(self):
+        """Thread is pinned to primary DB during deletion"""
+        pinned_states = []
+        original_submission_count = type(self.xform).submission_count
+
+        def capture_pinned(xform_self, force_update=False):
+            if force_update:
+                pinned_states.append(this_thread_is_pinned())
+            return original_submission_count(xform_self, force_update=force_update)
+
+        with patch.object(type(self.xform), "submission_count", capture_pinned):
+            delete_xform_submissions(self.xform, self.user)
+
+        self.assertTrue(
+            pinned_states, "submission_count was not called with force_update"
+        )
+        self.assertTrue(
+            all(pinned_states),
+            "Thread was not pinned to primary DB during submission_count",
+        )
+
 
 class ResponseWithMimetypeAndNameTestCase(TestBase):
     """Tests for method `response_with_mimetype_and_name`"""

onadata/libs/utils/logger_tools.py

@@ -1313,6 +1313,7 @@ def publish_xform(self):
         return publish_xml_form(self.xml_file, self.user, self.project)
 
 
+@use_master
 def delete_xform_submissions(
     xform: XForm,
     deleted_by: User,