diff --git a/openedx/core/djangoapps/content_libraries/api/blocks.py b/openedx/core/djangoapps/content_libraries/api/blocks.py
index dc0913d0fdc7..2a9a3790f48b 100644
--- a/openedx/core/djangoapps/content_libraries/api/blocks.py
+++ b/openedx/core/djangoapps/content_libraries/api/blocks.py
@@ -255,12 +255,12 @@ def set_library_block_olx(usage_key: LibraryUsageLocatorV2, new_olx_str: str) ->
 
     # .. event_implemented_name: LIBRARY_BLOCK_UPDATED
     # .. event_type: org.openedx.content_authoring.library_block.updated.v1
-    LIBRARY_BLOCK_UPDATED.send_event(
+    transaction.on_commit(lambda: LIBRARY_BLOCK_UPDATED.send_event(
         library_block=LibraryBlockData(
             library_key=usage_key.context_key,
             usage_key=usage_key
         )
-    )
+    ))
 
     # For each container, trigger LIBRARY_CONTAINER_UPDATED signal and set background=True to trigger
     # container indexing asynchronously.
@@ -268,12 +268,13 @@ def set_library_block_olx(usage_key: LibraryUsageLocatorV2, new_olx_str: str) ->
     for container in affected_containers:
         # .. event_implemented_name: LIBRARY_CONTAINER_UPDATED
         # .. event_type: org.openedx.content_authoring.content_library.container.updated.v1
-        LIBRARY_CONTAINER_UPDATED.send_event(
+        container_key = container.container_key
+        transaction.on_commit(lambda ck=container_key: LIBRARY_CONTAINER_UPDATED.send_event(  # type: ignore[misc]
             library_container=LibraryContainerData(
-                container_key=container.container_key,
+                container_key=ck,
                 background=True,
             )
-        )
+        ))
 
     return new_component_version
 
@@ -496,12 +497,12 @@ def _import_staged_block(
     # Emit library block created event
     # .. event_implemented_name: LIBRARY_BLOCK_CREATED
     # .. event_type: org.openedx.content_authoring.library_block.created.v1
-    LIBRARY_BLOCK_CREATED.send_event(
+    transaction.on_commit(lambda: LIBRARY_BLOCK_CREATED.send_event(
         library_block=LibraryBlockData(
             library_key=content_library.library_key,
             usage_key=usage_key
         )
-    )
+    ))
 
     # Now return the metadata about the new block
     return get_library_block(usage_key)
@@ -702,21 +703,34 @@ def delete_library_block(
     """
     Delete the specified block from this library (soft delete).
     """
-    component = get_component_from_usage_key(usage_key)
     library_key = usage_key.context_key
+
+    def send_block_deleted_signal():
+        # .. event_implemented_name: LIBRARY_BLOCK_DELETED
+        # .. event_type: org.openedx.content_authoring.library_block.deleted.v1
+        LIBRARY_BLOCK_DELETED.send_event(
+            library_block=LibraryBlockData(
+                library_key=library_key,
+                usage_key=usage_key
+            )
+        )
+
+    try:
+        component = get_component_from_usage_key(usage_key)
+    except Component.DoesNotExist:
+        # There may be cases where entries are created in the
+        # search index, but the component is not created
+        # (an intermediate error occurred).
+        # In that case, we keep the index updated by removing the entry.
+        send_block_deleted_signal()
+        return
+
     affected_collections = content_api.get_entity_collections(component.learning_package_id, component.key)
     affected_containers = get_containers_contains_item(usage_key)
 
     content_api.soft_delete_draft(component.pk, deleted_by=user_id)
 
-    # .. event_implemented_name: LIBRARY_BLOCK_DELETED
-    # .. event_type: org.openedx.content_authoring.library_block.deleted.v1
-    LIBRARY_BLOCK_DELETED.send_event(
-        library_block=LibraryBlockData(
-            library_key=library_key,
-            usage_key=usage_key
-        )
-    )
+    send_block_deleted_signal()
 
     # For each collection, trigger LIBRARY_COLLECTION_UPDATED signal and set background=True to trigger
     # collection indexing asynchronously.
diff --git a/openedx/core/djangoapps/content_libraries/api/containers.py b/openedx/core/djangoapps/content_libraries/api/containers.py
index 54740d130321..1751e038c8e2 100644
--- a/openedx/core/djangoapps/content_libraries/api/containers.py
+++ b/openedx/core/djangoapps/content_libraries/api/containers.py
@@ -8,6 +8,7 @@
 from uuid import uuid4
 import typing
 
+from django.db import transaction
 from django.utils.text import slugify
 from opaque_keys.edx.locator import LibraryContainerLocator, LibraryLocatorV2, LibraryUsageLocatorV2
 from openedx_events.content_authoring.data import (
@@ -154,11 +155,11 @@ def create_container(
 
     # .. event_implemented_name: LIBRARY_CONTAINER_CREATED
     # .. event_type: org.openedx.content_authoring.content_library.container.created.v1
-    LIBRARY_CONTAINER_CREATED.send_event(
+    transaction.on_commit(lambda: LIBRARY_CONTAINER_CREATED.send_event(
         library_container=LibraryContainerData(
             container_key=container_key,
         )
-    )
+    ))
 
     return ContainerMetadata.from_container(library_key, container)
 
@@ -261,8 +262,26 @@ def delete_container(
 
     No-op if container doesn't exist or has already been soft-deleted.
     """
+    def send_container_deleted_signal():
+        # .. event_implemented_name: LIBRARY_CONTAINER_DELETED
+        # .. event_type: org.openedx.content_authoring.content_library.container.deleted.v1
+        LIBRARY_CONTAINER_DELETED.send_event(
+            library_container=LibraryContainerData(
+                container_key=container_key,
+            )
+        )
+
+    try:
+        container = get_container_from_key(container_key)
+    except Container.DoesNotExist:
+        # There may be cases where entries are created in the
+        # search index, but the container is not created
+        # (an intermediate error occurred).
+        # In that case, we keep the index updated by removing the entry.
+        send_container_deleted_signal()
+        return
+
     library_key = container_key.lib_key
-    container = get_container_from_key(container_key)
 
     # Fetch related collections and containers before soft-delete
     affected_collections = content_api.get_entity_collections(
@@ -277,13 +296,7 @@ def delete_container(
     )
     content_api.soft_delete_draft(container.pk)
 
-    # .. event_implemented_name: LIBRARY_CONTAINER_DELETED
-    # .. event_type: org.openedx.content_authoring.content_library.container.deleted.v1
-    LIBRARY_CONTAINER_DELETED.send_event(
-        library_container=LibraryContainerData(
-            container_key=container_key,
-        )
-    )
+    send_container_deleted_signal()
 
     # For each collection, trigger LIBRARY_COLLECTION_UPDATED signal and set background=True to trigger
     # collection indexing asynchronously.
diff --git a/openedx/core/djangoapps/content_libraries/tests/test_api.py b/openedx/core/djangoapps/content_libraries/tests/test_api.py
index 69fae56b2d2c..b0aa5120708b 100644
--- a/openedx/core/djangoapps/content_libraries/tests/test_api.py
+++ b/openedx/core/djangoapps/content_libraries/tests/test_api.py
@@ -7,6 +7,7 @@
 import uuid
 from unittest import mock
 
+from django.db import transaction
 from django.test import TestCase
 from user_tasks.models import UserTaskStatus
 
@@ -18,16 +19,23 @@
 from opaque_keys.edx.locator import LibraryContainerLocator, LibraryLocatorV2, LibraryUsageLocatorV2
 from openedx_events.content_authoring.data import (
     ContentObjectChangedData,
+    LibraryBlockData,
     LibraryCollectionData,
     LibraryContainerData,
 )
 from openedx_events.content_authoring.signals import (
     CONTENT_OBJECT_ASSOCIATIONS_CHANGED,
+    LIBRARY_BLOCK_CREATED,
+    LIBRARY_BLOCK_DELETED,
+    LIBRARY_BLOCK_UPDATED,
     LIBRARY_COLLECTION_CREATED,
     LIBRARY_COLLECTION_DELETED,
     LIBRARY_COLLECTION_UPDATED,
+    LIBRARY_CONTAINER_CREATED,
+    LIBRARY_CONTAINER_DELETED,
     LIBRARY_CONTAINER_UPDATED,
 )
+from openedx_content.models_api import Component, Container
 from openedx_authz.api.users import get_user_role_assignments_in_scope
 from openedx_content import api as content_api
 
@@ -659,6 +667,69 @@ def test_delete_library_container(self) -> None:
             },
         )
 
+    def test_delete_container_when_container_does_not_exist(self) -> None:
+        """
+        Test that delete_container sends LIBRARY_CONTAINER_DELETED and returns
+        early (without calling soft_delete_draft) when the Container does not
+        exist in the DB. This covers the case where a search-index entry was
+        created but the container creation failed mid-way.
+        """
+        container_key = LibraryContainerLocator.from_string(self.unit1["id"])
+
+        event_receiver = mock.Mock()
+        LIBRARY_CONTAINER_DELETED.connect(event_receiver)
+        self.addCleanup(LIBRARY_CONTAINER_DELETED.disconnect, event_receiver)
+
+        with mock.patch(
+            "openedx.core.djangoapps.content_libraries.api.containers.get_container_from_key",
+            side_effect=Container.DoesNotExist,
+        ), mock.patch("openedx_content.api.soft_delete_draft") as mock_soft_delete:
+            api.delete_container(container_key)
+            mock_soft_delete.assert_not_called()
+
+        assert event_receiver.call_count == 1
+        self.assertDictContainsEntries(
+            event_receiver.call_args_list[0].kwargs,
+            {
+                "signal": LIBRARY_CONTAINER_DELETED,
+                "library_container": LibraryContainerData(
+                    container_key=container_key,
+                ),
+            },
+        )
+
+    def test_delete_library_block_when_component_does_not_exist(self) -> None:
+        """
+        Test that delete_library_block sends LIBRARY_BLOCK_DELETED and returns
+        early (without calling soft_delete_draft) when the Component does not
+        exist in the DB. This covers the case where a search-index entry was
+        created but the component creation failed mid-way.
+        """
+        usage_key = LibraryUsageLocatorV2.from_string(self.lib1_problem_block["id"])
+
+        event_receiver = mock.Mock()
+        LIBRARY_BLOCK_DELETED.connect(event_receiver)
+        self.addCleanup(LIBRARY_BLOCK_DELETED.disconnect, event_receiver)
+
+        with mock.patch(
+            "openedx.core.djangoapps.content_libraries.api.blocks.get_component_from_usage_key",
+            side_effect=Component.DoesNotExist,
+        ), mock.patch("openedx_content.api.soft_delete_draft") as mock_soft_delete:
+            api.delete_library_block(usage_key)
+            mock_soft_delete.assert_not_called()
+
+        assert event_receiver.call_count == 1
+        self.assertDictContainsEntries(
+            event_receiver.call_args_list[0].kwargs,
+            {
+                "signal": LIBRARY_BLOCK_DELETED,
+                "library_block": LibraryBlockData(
+                    library_key=self.lib1.library_key,
+                    usage_key=usage_key,
+                ),
+            },
+        )
+
     def test_restore_library_block(self) -> None:
         api.update_library_collection_items(
             self.lib1.library_key,
@@ -1390,6 +1461,101 @@ def test_copy_and_paste_container_another_library(self) -> None:
         # This is the same unit, so it should not be duplicated
         assert units_subsection1[0].container_key == units_subsection2[0].container_key
 
+    def test_set_library_block_olx_no_signal_on_rollback(self) -> None:
+        """
+        LIBRARY_BLOCK_UPDATED is NOT emitted when set_library_block_olx is called
+        within a transaction that is later rolled back.
+        """
+        event_receiver = mock.Mock()
+        LIBRARY_BLOCK_UPDATED.connect(event_receiver)
+        self.addCleanup(LIBRARY_BLOCK_UPDATED.disconnect, event_receiver)
+
+        try:
+            with transaction.atomic():
+                api.set_library_block_olx(
+                    self.problem_block_usage_key,
+                    "<problem>Updated inside rolled-back transaction</problem>",
+                )
+                raise RuntimeError("Force rollback")
+        except RuntimeError:
+            pass
+
+        assert event_receiver.call_count == 0
+
+    def test_set_library_block_olx_signal_emitted_on_success(self) -> None:
+        """
+        LIBRARY_BLOCK_UPDATED IS emitted when set_library_block_olx completes
+        successfully.
+        """
+        event_receiver = mock.Mock()
+        LIBRARY_BLOCK_UPDATED.connect(event_receiver)
+        self.addCleanup(LIBRARY_BLOCK_UPDATED.disconnect, event_receiver)
+
+        api.set_library_block_olx(
+            self.problem_block_usage_key,
+            "<problem>Updated successfully</problem>",
+        )
+
+        assert event_receiver.call_count == 1
+        self.assertDictContainsEntries(
+            event_receiver.call_args_list[0].kwargs,
+            {
+                "signal": LIBRARY_BLOCK_UPDATED,
+                "library_block": LibraryBlockData(
+                    library_key=self.lib1.library_key,
+                    usage_key=self.problem_block_usage_key,
+                ),
+            },
+        )
+
+    def test_import_container_no_signals_on_failure(self) -> None:
+        """
+        When import_staged_content_from_user_clipboard fails mid-way, none of
+        LIBRARY_CONTAINER_CREATED, LIBRARY_BLOCK_CREATED, or LIBRARY_BLOCK_UPDATED
+        are emitted, so the search index is not polluted with orphan entries.
+        """
+        api.copy_container(self.unit1.container_key, self.user.id)
+
+        event_receiver = mock.Mock()
+        for signal in [LIBRARY_CONTAINER_CREATED, LIBRARY_BLOCK_CREATED, LIBRARY_BLOCK_UPDATED]:
+            signal.connect(event_receiver)
+            self.addCleanup(signal.disconnect, event_receiver)
+
+        # Simulate a failure at the last step of the import (after the container
+        # and its child components have been created in the DB).
+        with mock.patch(
+            "openedx.core.djangoapps.content_libraries.api.blocks.update_container_children",
+            side_effect=RuntimeError("Simulated failure"),
+        ), self.assertRaises(RuntimeError):
+            api.import_staged_content_from_user_clipboard(self.lib1.library_key, self.user)
+
+        assert event_receiver.call_count == 0
+
+    def test_import_container_signals_emitted_on_success(self) -> None:
+        """
+        When import_staged_content_from_user_clipboard succeeds, LIBRARY_CONTAINER_CREATED
+        is emitted for the new container.
+        """
+        api.copy_container(self.unit1.container_key, self.user.id)
+
+        container_created_receiver = mock.Mock()
+        LIBRARY_CONTAINER_CREATED.connect(container_created_receiver)
+        self.addCleanup(LIBRARY_CONTAINER_CREATED.disconnect, container_created_receiver)
+
+        new_container = api.import_staged_content_from_user_clipboard(self.lib1.library_key, self.user)
+
+        assert container_created_receiver.call_count == 1
+        assert hasattr(new_container, "container_key")
+        self.assertDictContainsEntries(
+            container_created_receiver.call_args_list[0].kwargs,
+            {
+                "signal": LIBRARY_CONTAINER_CREATED,
+                "library_container": LibraryContainerData(
+                    container_key=new_container.container_key,  # type: ignore[union-attr]
+                ),
+            },
+        )
+
 
 class ContentLibraryExportTest(ContentLibrariesRestApiTest):
     """