fix: don't call signal handlers like XBLOCK_UPDATED before commit (#3…

…4800)

Co-authored-by: Yusuf Musleh <[email protected]>

xmodule/modulestore/__init__.py

@@ -121,6 +121,7 @@ def __init__(self):
         self._active_count = 0
         self.has_publish_item = False
         self.has_library_updated_item = False
+        self._commit_callbacks = []
 
     @property
     def active(self):
@@ -148,6 +149,20 @@ def is_root(self):
         """
         return self._active_count == 1
 
+    def defer_until_commit(self, fn):
+        """
+        Run some code when the changes from this bulk op are committed to the DB
+        """
+        self._commit_callbacks.append(fn)
+
+    def call_commit_callbacks(self):
+        """
+        When the changes have been committed to the DB, call this to run any queued callbacks
+        """
+        for fn in self._commit_callbacks:
+            fn()
+        self._commit_callbacks.clear()
+
 
 class ActiveBulkThread(threading.local):
     """
@@ -290,15 +305,34 @@ def _end_bulk_operation(self, structure_key, emit_signals=True, ignore_case=Fals
         # So re-nest until the signals are sent.
         bulk_ops_record.nest()
 
-        if emit_signals and dirty:
-            self.send_bulk_published_signal(bulk_ops_record, structure_key)
-            self.send_bulk_library_updated_signal(bulk_ops_record, structure_key)
+        if dirty:
+            # Call any "on commit" callback, regardless of if this was "published" or is still draft:
+            bulk_ops_record.call_commit_callbacks()
+            # Call any "on publish" handlers - emit_signals is usually false for draft-only changes:
+            if emit_signals:
+                self.send_bulk_published_signal(bulk_ops_record, structure_key)
+                self.send_bulk_library_updated_signal(bulk_ops_record, structure_key)
 
         # Signals are sent. Now unnest and clear the bulk op for good.
         bulk_ops_record.unnest()
 
         self._clear_bulk_ops_record(structure_key)
 
+    def on_commit_changes_to(self, course_key, fn):
+        """
+        Call some callback when the currently active bulk operation has saved
+        """
+        # Check if a bulk op is active. If so, defer fn(); otherwise call it immediately.
+        # Note: calling _get_bulk_ops_record() here and then checking .active can have side-effects in some cases
+        # because it creates an entry in the defaultdict if none exists, so we check if the record is active using
+        # the same code as _clear_bulk_ops_record(), which doesn't modify the defaultdict.
+        # so we check it this way:
+        if course_key and course_key.for_branch(None) in self._active_bulk_ops.records:
+            bulk_ops_record = self._active_bulk_ops.records[course_key.for_branch(None)]
+            bulk_ops_record.defer_until_commit(fn)
+        else:
+            fn()  # There is no active bulk operation - call fn() now.
+
     def _is_in_bulk_operation(self, course_key, ignore_case=False):
         """
         Return whether a bulk operation is active on `course_key`.

xmodule/modulestore/mixed.py

@@ -758,15 +758,19 @@ def create_item(self, user_id, course_key, block_type, block_id=None, fields=Non
         """
         modulestore = self._verify_modulestore_support(course_key, 'create_item')
         xblock = modulestore.create_item(user_id, course_key, block_type, block_id=block_id, fields=fields, **kwargs)
-        # .. event_implemented_name: XBLOCK_CREATED
-        XBLOCK_CREATED.send_event(
-            time=datetime.now(timezone.utc),
-            xblock_info=XBlockData(
-                usage_key=xblock.location.for_branch(None),
-                block_type=block_type,
-                version=xblock.location
+
+        def send_created_event():
+            # .. event_implemented_name: XBLOCK_CREATED
+            XBLOCK_CREATED.send_event(
+                time=datetime.now(timezone.utc),
+                xblock_info=XBlockData(
+                    usage_key=xblock.location.for_branch(None),
+                    block_type=block_type,
+                    version=xblock.location
+                )
             )
-        )
+
+        modulestore.on_commit_changes_to(course_key, send_created_event)
         return xblock
 
     @strip_key
@@ -787,19 +791,24 @@ def create_child(self, user_id, parent_usage_key, block_type, block_id=None, fie
             fields (dict): A dictionary specifying initial values for some or all fields
                 in the newly created block
         """
-        modulestore = self._verify_modulestore_support(parent_usage_key.course_key, 'create_child')
-        xblock = modulestore.create_child(
+        course_key = parent_usage_key.course_key
+        store = self._verify_modulestore_support(course_key, 'create_child')
+        xblock = store.create_child(
             user_id, parent_usage_key, block_type, block_id=block_id, fields=fields, **kwargs
         )
-        # .. event_implemented_name: XBLOCK_CREATED
-        XBLOCK_CREATED.send_event(
-            time=datetime.now(timezone.utc),
-            xblock_info=XBlockData(
-                usage_key=xblock.location.for_branch(None),
-                block_type=block_type,
-                version=xblock.location
+
+        def send_created_event():
+            # .. event_implemented_name: XBLOCK_CREATED
+            XBLOCK_CREATED.send_event(
+                time=datetime.now(timezone.utc),
+                xblock_info=XBlockData(
+                    usage_key=xblock.location.for_branch(None),
+                    block_type=block_type,
+                    version=xblock.location
+                )
             )
-        )
+
+        store.on_commit_changes_to(course_key, send_created_event)
         return xblock
 
     @strip_key
@@ -828,34 +837,44 @@ def update_item(self, xblock, user_id, allow_not_found=False, **kwargs):  # lint
         Update the xblock persisted to be the same as the given for all types of fields
         (content, children, and metadata) attribute the change to the given user.
         """
-        store = self._verify_modulestore_support(xblock.location.course_key, 'update_item')
+        course_key = xblock.location.course_key
+        store = self._verify_modulestore_support(course_key, 'update_item')
         xblock = store.update_item(xblock, user_id, allow_not_found, **kwargs)
-        # .. event_implemented_name: XBLOCK_UPDATED
-        XBLOCK_UPDATED.send_event(
-            time=datetime.now(timezone.utc),
-            xblock_info=XBlockData(
-                usage_key=xblock.location.for_branch(None),
-                block_type=xblock.location.block_type,
-                version=xblock.location
+
+        def send_updated_event():
+            # .. event_implemented_name: XBLOCK_UPDATED
+            XBLOCK_UPDATED.send_event(
+                time=datetime.now(timezone.utc),
+                xblock_info=XBlockData(
+                    usage_key=xblock.location.for_branch(None),
+                    block_type=xblock.location.block_type,
+                    version=xblock.location
+                )
             )
-        )
+
+        store.on_commit_changes_to(course_key, send_updated_event)
         return xblock
 
     @strip_key
     def delete_item(self, location, user_id, **kwargs):  # lint-amnesty, pylint: disable=arguments-differ
         """
         Delete the given item from persistence. kwargs allow modulestore specific parameters.
         """
-        store = self._verify_modulestore_support(location.course_key, 'delete_item')
+        course_key = location.course_key
+        store = self._verify_modulestore_support(course_key, 'delete_item')
         item = store.delete_item(location, user_id=user_id, **kwargs)
-        # .. event_implemented_name: XBLOCK_DELETED
-        XBLOCK_DELETED.send_event(
-            time=datetime.now(timezone.utc),
-            xblock_info=XBlockData(
-                usage_key=location,
-                block_type=location.block_type,
+
+        def send_deleted_event():
+            # .. event_implemented_name: XBLOCK_DELETED
+            XBLOCK_DELETED.send_event(
+                time=datetime.now(timezone.utc),
+                xblock_info=XBlockData(
+                    usage_key=location,
+                    block_type=location.block_type,
+                )
             )
-        )
+
+        store.on_commit_changes_to(course_key, send_deleted_event)
         return item
 
     def revert_to_published(self, location, user_id):