fix: schedule tasks with on_commit

Author: aaxelb

share/models/index_backfill.py

@@ -98,14 +98,11 @@ def pls_start(self, index_strategy):
                 locked_self.strategy_checksum = _current_checksum
                 locked_self.backfill_status = IndexBackfill.INITIAL
             locked_self.__update_error(None)
-            try:
-                task__schedule_index_backfill.apply_async((locked_self.pk,))
-            except Exception as error:
-                locked_self.__update_error(error)
-            else:
-                locked_self.backfill_status = IndexBackfill.WAITING
-            finally:
-                locked_self.save()
+            locked_self.backfill_status = IndexBackfill.WAITING
+            locked_self.save()
+            transaction.on_commit(
+                lambda: task__schedule_index_backfill.apply_async((locked_self.pk,))
+            )
 
     def pls_note_scheduling_has_begun(self):
         with self.mutex() as locked_self:

tests/share/search/test_index_backfill.py

@@ -20,12 +20,15 @@ def index_backfill(self, fake_strategy):
             index_strategy_name=fake_strategy.strategy_name,
         )
 
-    def test_happypath(self, index_backfill: IndexBackfill, fake_strategy):
+    def test_happypath(self, index_backfill: IndexBackfill, fake_strategy, django_capture_on_commit_callbacks):
         assert index_backfill.backfill_status == IndexBackfill.INITIAL
         assert index_backfill.strategy_checksum == ''
-        with mock.patch('share.models.index_backfill.task__schedule_index_backfill') as mock_task:
+        with (
+            mock.patch('share.models.index_backfill.task__schedule_index_backfill') as mock_task,
+            django_capture_on_commit_callbacks(execute=True),
+        ):
             index_backfill.pls_start(fake_strategy)
-            mock_task.apply_async.assert_called_once_with((index_backfill.pk,))
+        mock_task.apply_async.assert_called_once_with((index_backfill.pk,))
         assert index_backfill.backfill_status == IndexBackfill.WAITING
         assert index_backfill.strategy_checksum == 'foo_bar'
         index_backfill.pls_note_scheduling_has_begun()

tests/trove/digestive_tract/test_expel.py

@@ -61,11 +61,13 @@ def test_setup(self):
     def test_expel(self):
         with mock.patch('trove.digestive_tract.expel_suid') as _mock_expel_suid:
             _user = self.suid_1.source_config.source.user
-            digestive_tract.expel(from_user=_user, record_identifier=self.suid_1.identifier)
+            with self.captureOnCommitCallbacks(execute=True):
+                digestive_tract.expel(from_user=_user, record_identifier=self.suid_1.identifier)
         _mock_expel_suid.assert_called_once_with(self.suid_1)
 
     def test_expel_suid(self):
-        digestive_tract.expel_suid(self.suid_1)
+        with self.captureOnCommitCallbacks(execute=True):
+            digestive_tract.expel_suid(self.suid_1)
         self.indexcard_1.refresh_from_db()
         self.indexcard_2.refresh_from_db()
         self.assertIsNotNone(self.indexcard_1.deleted)
@@ -85,7 +87,8 @@ def test_expel_suid(self):
         self.mock_derive_task.delay.assert_not_called()
 
     def test_expel_supplementary_suid(self):
-        digestive_tract.expel_suid(self.supp_suid)
+        with self.captureOnCommitCallbacks(execute=True):
+            digestive_tract.expel_suid(self.supp_suid)
         self.indexcard_1.refresh_from_db()
         self.indexcard_2.refresh_from_db()
         self.assertIsNone(self.indexcard_1.deleted)
@@ -105,15 +108,17 @@ def test_expel_supplementary_suid(self):
 
     def test_expel_expired_task(self):
         with mock.patch('trove.digestive_tract.expel_expired_data') as _mock_expel_expired:
-            digestive_tract.task__expel_expired_data.apply()
+            with self.captureOnCommitCallbacks(execute=True):
+                digestive_tract.task__expel_expired_data.apply()
         _mock_expel_expired.assert_called_once_with(datetime.date.today())
 
     def test_expel_expired(self):
         _today = datetime.date.today()
         _latest = self.indexcard_2.latest_resource_description
         _latest.expiration_date = _today
         _latest.save()
-        digestive_tract.expel_expired_data(_today)
+        with self.captureOnCommitCallbacks(execute=True):
+            digestive_tract.expel_expired_data(_today)
         self.indexcard_1.refresh_from_db()
         self.indexcard_2.refresh_from_db()
         self.assertIsNone(self.indexcard_1.deleted)
@@ -136,7 +141,8 @@ def test_expel_expired_supplement(self):
         _today = datetime.date.today()
         self.supp.expiration_date = _today
         self.supp.save()
-        digestive_tract.expel_expired_data(_today)
+        with self.captureOnCommitCallbacks(execute=True):
+            digestive_tract.expel_expired_data(_today)
         self.indexcard_1.refresh_from_db()
         self.indexcard_2.refresh_from_db()
         self.assertIsNone(self.indexcard_1.deleted)

trove/digestive_tract.py

@@ -65,7 +65,7 @@ def ingest(
             expiration_date=expiration_date,
         )
         for _card in _extracted_cards:
-            task__derive.delay(_card.pk, urgent=urgent)
+            transaction.on_commit(lambda: task__derive.delay(_card.pk, urgent=urgent))
 
 
 @transaction.atomic
@@ -249,7 +249,7 @@ def _expel_supplementary_descriptions(supplementary_rdf_queryset: QuerySet[trove
             _affected_indexcards.add(_supplement.indexcard)
         _supplement.delete()
     for _indexcard in _affected_indexcards:
-        task__derive.delay(_indexcard.pk)
+        transaction.on_commit(lambda: task__derive.delay(_indexcard.pk))
 
 
 ### BEGIN celery tasks