diff --git a/cms/djangoapps/contentstore/management/commands/import.py b/cms/djangoapps/contentstore/management/commands/import.py
index 4ff9e7ce8d43..346f10d933b1 100644
--- a/cms/djangoapps/contentstore/management/commands/import.py
+++ b/cms/djangoapps/contentstore/management/commands/import.py
@@ -8,7 +8,7 @@
 from openedx.core.djangoapps.django_comment_common.utils import are_permissions_roles_seeded, seed_permissions_roles
 from xmodule.contentstore.django import contentstore  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.modulestore import ModuleStoreEnum  # lint-amnesty, pylint: disable=wrong-import-order
-from xmodule.modulestore.django import modulestore  # lint-amnesty, pylint: disable=wrong-import-order
+from xmodule.modulestore.django import SignalHandler, modulestore  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.modulestore.xml_importer import import_course_from_xml  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.util.sandboxing import DEFAULT_PYTHON_LIB_FILENAME  # lint-amnesty, pylint: disable=wrong-import-order
 
@@ -73,6 +73,10 @@ def handle(self, *args, **options):
 
         for course in course_items:
             course_id = course.id
+            # Importing is an act of publishing so send the course published signal.
+            SignalHandler.course_published.send_robust(sender=self, course_key=course_id)
+
+            # Seed forum permission roles if we need to.
             if not are_permissions_roles_seeded(course_id):
                 self.stdout.write(f'Seeding forum roles for course {course_id}\n')
                 seed_permissions_roles(course_id)
diff --git a/lms/djangoapps/ccx/tests/test_field_override_performance.py b/lms/djangoapps/ccx/tests/test_field_override_performance.py
index 68926a24f2dd..a7128495eb59 100644
--- a/lms/djangoapps/ccx/tests/test_field_override_performance.py
+++ b/lms/djangoapps/ccx/tests/test_field_override_performance.py
@@ -234,7 +234,7 @@ class TestFieldOverrideSplitPerformance(FieldOverridePerformanceTestCase):
     __test__ = True
 
     # TODO: decrease query count as part of REVO-28
-    QUERY_COUNT = 33
+    QUERY_COUNT = 34
 
     TEST_DATA = {
         ('no_overrides', 1, True, False): (QUERY_COUNT, 2),
diff --git a/lms/djangoapps/course_api/blocks/tests/test_api.py b/lms/djangoapps/course_api/blocks/tests/test_api.py
index e6fd74463aa9..93e400d4a804 100644
--- a/lms/djangoapps/course_api/blocks/tests/test_api.py
+++ b/lms/djangoapps/course_api/blocks/tests/test_api.py
@@ -3,16 +3,13 @@
 """
 
 
-from itertools import product
 from unittest.mock import patch
 
 import ddt
 from django.test.client import RequestFactory
-from edx_toggles.toggles.testutils import override_waffle_switch
 
 from common.djangoapps.student.tests.factories import UserFactory
 from openedx.core.djangoapps.content.block_structure.api import clear_course_from_cache
-from openedx.core.djangoapps.content.block_structure.config import STORAGE_BACKING_FOR_CACHE
 from xmodule.modulestore import ModuleStoreEnum  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.modulestore.tests.factories import SampleCourseFactory, check_mongo_calls  # lint-amnesty, pylint: disable=wrong-import-order
@@ -209,34 +206,25 @@ class TestGetBlocksQueryCounts(TestGetBlocksQueryCountsBase):
     Tests query counts for the get_blocks function.
     """
 
-    @ddt.data(
-        *product(
-            (ModuleStoreEnum.Type.split, ),
-            (True, False),
+    @ddt.data(ModuleStoreEnum.Type.split)
+    def test_query_counts_cached(self, store_type):
+        course = self._create_course(store_type)
+        self._get_blocks(
+            course,
+            expected_mongo_queries=0,
+            expected_sql_queries=14,
         )
-    )
-    @ddt.unpack
-    def test_query_counts_cached(self, store_type, with_storage_backing):
-        with override_waffle_switch(STORAGE_BACKING_FOR_CACHE, active=with_storage_backing):
-            course = self._create_course(store_type)
-            self._get_blocks(
-                course,
-                expected_mongo_queries=0,
-                expected_sql_queries=14 if with_storage_backing else 13,
-            )
 
     @ddt.data(
-        (ModuleStoreEnum.Type.split, 2, True, 24),
-        (ModuleStoreEnum.Type.split, 2, False, 14),
+        (ModuleStoreEnum.Type.split, 2, 24),
     )
     @ddt.unpack
-    def test_query_counts_uncached(self, store_type, expected_mongo_queries, with_storage_backing, num_sql_queries):
-        with override_waffle_switch(STORAGE_BACKING_FOR_CACHE, active=with_storage_backing):
-            course = self._create_course(store_type)
-            clear_course_from_cache(course.id)
-
-            self._get_blocks(
-                course,
-                expected_mongo_queries,
-                expected_sql_queries=num_sql_queries,
-            )
+    def test_query_counts_uncached(self, store_type, expected_mongo_queries, num_sql_queries):
+        course = self._create_course(store_type)
+        clear_course_from_cache(course.id)
+
+        self._get_blocks(
+            course,
+            expected_mongo_queries,
+            expected_sql_queries=num_sql_queries,
+        )
diff --git a/lms/djangoapps/course_api/blocks/transformers/tests/test_milestones.py b/lms/djangoapps/course_api/blocks/transformers/tests/test_milestones.py
index e87e0e755685..5efea79c1986 100644
--- a/lms/djangoapps/course_api/blocks/transformers/tests/test_milestones.py
+++ b/lms/djangoapps/course_api/blocks/transformers/tests/test_milestones.py
@@ -13,6 +13,7 @@
 from lms.djangoapps.course_blocks.api import get_course_blocks
 from lms.djangoapps.course_blocks.transformers.tests.helpers import CourseStructureTestCase
 from lms.djangoapps.gating import api as lms_gating_api
+import openedx.core.djangoapps.content.block_structure.api as bs_api
 from openedx.core.djangoapps.content.block_structure.transformers import BlockStructureTransformers
 from openedx.core.djangoapps.course_apps.toggles import EXAMS_IDA
 from openedx.core.lib.gating import api as gating_api
@@ -166,7 +167,11 @@ def test_gated(self, gated_block_ref, gating_block_ref, expected_blocks_before_c
         self.course.enable_subsection_gating = True
         self.setup_gated_section(self.blocks[gated_block_ref], self.blocks[gating_block_ref])
 
-        with self.assertNumQueries(5):
+        # Cache the course blocks so that they don't need to be generated when we're trying to
+        # get data back.  This would happen as a part of publishing in a production system.
+        bs_api.update_course_in_cache(self.course.id)
+
+        with self.assertNumQueries(4):
             self.get_blocks_and_check_against_expected(self.user, expected_blocks_before_completion)
 
         # clear the request cache to simulate a new request
@@ -179,7 +184,7 @@ def test_gated(self, gated_block_ref, gating_block_ref, expected_blocks_before_c
             self.user,
         )
 
-        with self.assertNumQueries(6):
+        with self.assertNumQueries(4):
             self.get_blocks_and_check_against_expected(self.user, self.ALL_BLOCKS_EXCEPT_SPECIAL)
 
     def test_staff_access(self):
diff --git a/lms/djangoapps/course_blocks/transformers/tests/test_library_content.py b/lms/djangoapps/course_blocks/transformers/tests/test_library_content.py
index 7fdb58f7f612..5a4d7a0de11a 100644
--- a/lms/djangoapps/course_blocks/transformers/tests/test_library_content.py
+++ b/lms/djangoapps/course_blocks/transformers/tests/test_library_content.py
@@ -8,6 +8,7 @@
 from openedx.core.djangoapps.content.block_structure.api import clear_course_from_cache
 from openedx.core.djangoapps.content.block_structure.transformers import BlockStructureTransformers
 
+import openedx.core.djangoapps.content.block_structure.api as bs_api
 from ...api import get_course_blocks
 from ..library_content import ContentLibraryOrderTransformer, ContentLibraryTransformer
 from .helpers import CourseStructureTestCase
@@ -41,6 +42,8 @@ def setUp(self):
         self.course_hierarchy = self.get_course_hierarchy()
         self.blocks = self.build_course(self.course_hierarchy)
         self.course = self.blocks['course']
+        # Do this manually because publish signals are not fired by default in tests.
+        bs_api.update_course_in_cache(self.course.id)
         clear_course_from_cache(self.course.id)
 
         # Enroll user in course.
@@ -122,6 +125,7 @@ def test_content_library(self):
         )
         assert len(list(raw_block_structure.get_block_keys())) == len(self.blocks)
 
+        bs_api.update_course_in_cache(self.course.id)
         clear_course_from_cache(self.course.id)
         trans_block_structure = get_course_blocks(
             self.user,
@@ -175,6 +179,7 @@ def setUp(self):
         self.course_hierarchy = self.get_course_hierarchy()
         self.blocks = self.build_course(self.course_hierarchy)
         self.course = self.blocks['course']
+        bs_api.update_course_in_cache(self.course.id)
         clear_course_from_cache(self.course.id)
 
         # Enroll user in course.
diff --git a/lms/djangoapps/courseware/tests/test_views.py b/lms/djangoapps/courseware/tests/test_views.py
index d38110b3b045..126233880ae1 100644
--- a/lms/djangoapps/courseware/tests/test_views.py
+++ b/lms/djangoapps/courseware/tests/test_views.py
@@ -1472,8 +1472,8 @@ def test_view_certificate_link(self):
             self.assertContains(resp, "earned a certificate for this course.")
 
     @ddt.data(
-        (True, 53),
-        (False, 53),
+        (True, 54),
+        (False, 54),
     )
     @ddt.unpack
     def test_progress_queries_paced_courses(self, self_paced, query_count):
@@ -1488,13 +1488,13 @@ def test_progress_queries(self):
         ContentTypeGatingConfig.objects.create(enabled=True, enabled_as_of=datetime(2018, 1, 1))
         self.setup_course()
         with self.assertNumQueries(
-            53, table_ignorelist=QUERY_COUNT_TABLE_IGNORELIST
+            54, table_ignorelist=QUERY_COUNT_TABLE_IGNORELIST
         ), check_mongo_calls(2):
             self._get_progress_page()
 
         for _ in range(2):
             with self.assertNumQueries(
-                38, table_ignorelist=QUERY_COUNT_TABLE_IGNORELIST
+                39, table_ignorelist=QUERY_COUNT_TABLE_IGNORELIST
             ), check_mongo_calls(2):
                 self._get_progress_page()
 
diff --git a/lms/djangoapps/grades/rest_api/v1/tests/test_gradebook_views.py b/lms/djangoapps/grades/rest_api/v1/tests/test_gradebook_views.py
index 5d7c428c2221..42dd9139ee21 100644
--- a/lms/djangoapps/grades/rest_api/v1/tests/test_gradebook_views.py
+++ b/lms/djangoapps/grades/rest_api/v1/tests/test_gradebook_views.py
@@ -22,6 +22,7 @@
 from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory, BlockFactory
 
+import openedx.core.djangoapps.content.block_structure.api as bs_api
 from common.djangoapps.course_modes.models import CourseMode
 from common.djangoapps.student.roles import (
     CourseBetaTesterRole,
@@ -2228,6 +2229,9 @@ def test_get_override_for_unreleased_block(self):
             display_name='Unreleased Section',
         )
 
+        # We need to update the course in the cache after we create the new block.
+        bs_api.update_course_in_cache(self.course_data.course_key)
+
         resp = self.client.get(
             self.get_url(subsection_id=unreleased_subsection.location)
         )
diff --git a/lms/djangoapps/grades/tests/integration/test_events.py b/lms/djangoapps/grades/tests/integration/test_events.py
index f058cc81799e..315ffd65ef25 100644
--- a/lms/djangoapps/grades/tests/integration/test_events.py
+++ b/lms/djangoapps/grades/tests/integration/test_events.py
@@ -7,6 +7,7 @@
 
 from crum import set_current_request
 
+import openedx.core.djangoapps.content.block_structure.api as bs_api
 from xmodule.capa.tests.response_xml_factory import MultipleChoiceResponseXMLFactory
 from common.djangoapps.student.models import CourseEnrollment
 from common.djangoapps.student.tests.factories import UserFactory
@@ -76,44 +77,45 @@ def setUp(self):
         CourseEnrollment.enroll(self.student, self.course.id)
         self.instructor = UserFactory.create(is_staff=True, username='test_instructor', password=self.TEST_PASSWORD)
         self.refresh_course()
+        # Since this doesn't happen automatically and we don't want to run all the publish signal handlers
+        # Just make sure we have the latest version of the course in cache before we test the problem.
+        bs_api.update_course_in_cache(self.course.id)
 
     @patch('lms.djangoapps.grades.events.tracker')
     def test_submit_answer(self, events_tracker):
         self.submit_question_answer('p1', {'2_1': 'choice_choice_2'})
-        course = self.store.get_course(self.course.id, depth=0)
 
         event_transaction_id = events_tracker.emit.mock_calls[0][1][1]['event_transaction_id']
-        events_tracker.emit.assert_has_calls(
-            [
-                mock_call(
-                    events.PROBLEM_SUBMITTED_EVENT_TYPE,
-                    {
-                        'user_id': str(self.student.id),
-                        'event_transaction_id': event_transaction_id,
-                        'event_transaction_type': events.PROBLEM_SUBMITTED_EVENT_TYPE,
-                        'course_id': str(self.course.id),
-                        'problem_id': str(self.problem.location),
-                        'weighted_earned': 2.0,
-                        'weighted_possible': 2.0,
-                    },
-                ),
-                mock_call(
-                    events.COURSE_GRADE_CALCULATED,
-                    {
-                        'course_version': str(course.course_version),
-                        'percent_grade': 0.02,
-                        'grading_policy_hash': 'ChVp0lHGQGCevD0t4njna/C44zQ=',
-                        'user_id': str(self.student.id),
-                        'letter_grade': '',
-                        'event_transaction_id': event_transaction_id,
-                        'event_transaction_type': events.PROBLEM_SUBMITTED_EVENT_TYPE,
-                        'course_id': str(self.course.id),
-                        'course_edited_timestamp': str(course.subtree_edited_on),
-                    }
-                ),
-            ],
-            any_order=True,
-        )
+        expected_calls = [
+            mock_call(
+                events.PROBLEM_SUBMITTED_EVENT_TYPE,
+                {
+                    'user_id': str(self.student.id),
+                    'event_transaction_id': event_transaction_id,
+                    'event_transaction_type': events.PROBLEM_SUBMITTED_EVENT_TYPE,
+                    'course_id': str(self.course.id),
+                    'problem_id': str(self.problem.location),
+                    'weighted_earned': 2.0,
+                    'weighted_possible': 2.0,
+                },
+            ),
+            mock_call(
+                events.COURSE_GRADE_CALCULATED,
+                {
+                    'course_version': str(self.course.course_version),
+                    'percent_grade': 0.02,
+                    'grading_policy_hash': 'ChVp0lHGQGCevD0t4njna/C44zQ=',
+                    'user_id': str(self.student.id),
+                    'letter_grade': '',
+                    'event_transaction_id': event_transaction_id,
+                    'event_transaction_type': events.PROBLEM_SUBMITTED_EVENT_TYPE,
+                    'course_id': str(self.course.id),
+                    'course_edited_timestamp': str(self.course.subtree_edited_on),
+                }
+            ),
+        ]
+
+        events_tracker.emit.assert_has_calls(expected_calls, any_order=True)
 
     @ddt.data(True, False)
     def test_delete_student_state(self, emit_signals):
diff --git a/lms/djangoapps/grades/tests/test_course_grade_factory.py b/lms/djangoapps/grades/tests/test_course_grade_factory.py
index 2066b6cd0d7c..c47e80a3dae0 100644
--- a/lms/djangoapps/grades/tests/test_course_grade_factory.py
+++ b/lms/djangoapps/grades/tests/test_course_grade_factory.py
@@ -68,35 +68,35 @@ def _assert_section_order(course_grade):
                 self.sequence2.display_name
             ]
 
-        with self.assertNumQueries(4), mock_get_score(1, 2):
+        with self.assertNumQueries(5), mock_get_score(1, 2):
             _assert_read(expected_pass=False, expected_percent=0)  # start off with grade of 0
 
-        num_queries = 42
+        num_queries = 43
         with self.assertNumQueries(num_queries), mock_get_score(1, 2):
             grade_factory.update(self.request.user, self.course, force_update_subsections=True)
 
-        with self.assertNumQueries(3):
+        with self.assertNumQueries(4):
             _assert_read(expected_pass=True, expected_percent=0.5)  # updated to grade of .5
 
-        num_queries = 6
+        num_queries = 7
         with self.assertNumQueries(num_queries), mock_get_score(1, 4):
             grade_factory.update(self.request.user, self.course, force_update_subsections=False)
 
-        with self.assertNumQueries(3):
+        with self.assertNumQueries(4):
             _assert_read(expected_pass=True, expected_percent=0.5)  # NOT updated to grade of .25
 
-        num_queries = 18
+        num_queries = 19
         with self.assertNumQueries(num_queries), mock_get_score(2, 2):
             grade_factory.update(self.request.user, self.course, force_update_subsections=True)
 
-        with self.assertNumQueries(3):
+        with self.assertNumQueries(4):
             _assert_read(expected_pass=True, expected_percent=1.0)  # updated to grade of 1.0
 
-        num_queries = 28
+        num_queries = 29
         with self.assertNumQueries(num_queries), mock_get_score(0, 0):  # the subsection now is worth zero
             grade_factory.update(self.request.user, self.course, force_update_subsections=True)
 
-        with self.assertNumQueries(3):
+        with self.assertNumQueries(4):
             _assert_read(expected_pass=False, expected_percent=0.0)  # updated to grade of 0.0
 
     @ddt.data((True, False))
@@ -286,7 +286,7 @@ def test_grading_exception(self, mock_course_grade):
             else mock_course_grade.return_value
             for student in self.students
         ]
-        with self.assertNumQueries(11):
+        with self.assertNumQueries(20):
             all_course_grades, all_errors = self._course_grades_and_errors_for(self.course, self.students)
         assert {student: str(all_errors[student]) for student in all_errors} == {
             student3: 'Error for student3.',
diff --git a/lms/djangoapps/grades/tests/test_tasks.py b/lms/djangoapps/grades/tests/test_tasks.py
index 4dfd2dbef97f..347879105aad 100644
--- a/lms/djangoapps/grades/tests/test_tasks.py
+++ b/lms/djangoapps/grades/tests/test_tasks.py
@@ -156,8 +156,8 @@ def test_block_structure_created_only_once(self):
             assert mock_block_structure_create.call_count == 1
 
     @ddt.data(
-        (ModuleStoreEnum.Type.split, 2, 41, True),
-        (ModuleStoreEnum.Type.split, 2, 41, False),
+        (ModuleStoreEnum.Type.split, 2, 42, True),
+        (ModuleStoreEnum.Type.split, 2, 42, False),
     )
     @ddt.unpack
     def test_query_counts(self, default_store, num_mongo_calls, num_sql_calls, create_multiple_subsections):
@@ -168,7 +168,7 @@ def test_query_counts(self, default_store, num_mongo_calls, num_sql_calls, creat
                     self._apply_recalculate_subsection_grade()
 
     @ddt.data(
-        (ModuleStoreEnum.Type.split, 2, 41),
+        (ModuleStoreEnum.Type.split, 2, 42),
     )
     @ddt.unpack
     def test_query_counts_dont_change_with_more_content(self, default_store, num_mongo_calls, num_sql_calls):
@@ -255,7 +255,7 @@ def test_problem_block_with_restricted_access(self, mock_subsection_signal):
         UserPartition.scheme_extensions = None
 
     @ddt.data(
-        (ModuleStoreEnum.Type.split, 2, 41),
+        (ModuleStoreEnum.Type.split, 2, 42),
     )
     @ddt.unpack
     def test_persistent_grades_on_course(self, default_store, num_mongo_queries, num_sql_queries):
diff --git a/lms/djangoapps/grades/tests/test_transformer.py b/lms/djangoapps/grades/tests/test_transformer.py
index b0cb6c0d49fb..7ef13e5b5b7c 100644
--- a/lms/djangoapps/grades/tests/test_transformer.py
+++ b/lms/djangoapps/grades/tests/test_transformer.py
@@ -13,6 +13,7 @@
 from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
 from xmodule.modulestore.tests.factories import check_mongo_calls_range
 
+import openedx.core.djangoapps.content.block_structure.api as bs_api
 from common.djangoapps.student.tests.factories import UserFactory
 from lms.djangoapps.course_blocks.api import get_course_blocks
 from lms.djangoapps.course_blocks.transformers.tests.helpers import CourseStructureTestCase
@@ -462,6 +463,7 @@ def test_modulestore_performance(self, store_type, max_mongo_calls, min_mongo_ca
             )
         with self.store.default_store(store_type):
             blocks = self.build_course(course)
+        bs_api.update_course_in_cache(blocks['course'].id)
         clear_course_from_cache(blocks['course'].id)
         with check_mongo_calls_range(max_mongo_calls, min_mongo_calls):
             get_course_blocks(self.student, blocks['course'].location, self.transformers)
diff --git a/lms/djangoapps/instructor_task/tests/test_tasks_helper.py b/lms/djangoapps/instructor_task/tests/test_tasks_helper.py
index 1fb25aeb8c07..0a8c5e788bfc 100644
--- a/lms/djangoapps/instructor_task/tests/test_tasks_helper.py
+++ b/lms/djangoapps/instructor_task/tests/test_tasks_helper.py
@@ -25,6 +25,7 @@
 from pytz import UTC
 
 import openedx.core.djangoapps.user_api.course_tag.api as course_tag_api
+import openedx.core.djangoapps.content.block_structure.api as bs_api
 from xmodule.capa.tests.response_xml_factory import MultipleChoiceResponseXMLFactory  # lint-amnesty, pylint: disable=wrong-import-order
 from common.djangoapps.course_modes.models import CourseMode
 from common.djangoapps.student.models import CourseEnrollment, CourseEnrollmentAllowed
@@ -396,6 +397,8 @@ def test_query_counts(self):
         )
         _ = CreditCourseFactory(course_key=course.id)
 
+        bs_api.update_course_in_cache(course.id)
+
         num_users = 5
         for _ in range(num_users):
             user = UserFactory.create()
@@ -406,7 +409,7 @@ def test_query_counts(self):
 
         with patch('lms.djangoapps.instructor_task.tasks_helper.runner._get_current_task'):
             with check_mongo_calls(2):
-                with self.assertNumQueries(54):
+                with self.assertNumQueries(46):
                     CourseGradeReport.generate(None, None, course.id, {}, 'graded')
 
     def test_inactive_enrollments(self):
diff --git a/openedx/core/djangoapps/content/block_structure/__init__.py b/openedx/core/djangoapps/content/block_structure/__init__.py
index f77cf27f041f..00ed1b12d2ef 100644
--- a/openedx/core/djangoapps/content/block_structure/__init__.py
+++ b/openedx/core/djangoapps/content/block_structure/__init__.py
@@ -59,4 +59,10 @@
 Note: A partial subset (as an ordered list) of the registered
 transformers can be requested during the Transform phase, allowing
 the client to manipulate exactly which transformers to call.
+
+Links to Other Block Structure Related Documentation:
+
+  * https://openedx.atlassian.net/wiki/spaces/AC/pages/154861855/Block+Structure+Cache+Invalidation+Proposal
+  * https://openedx.atlassian.net/wiki/spaces/AC/pages/34734111/Course+Block+Transformers
+  * https://openedx.atlassian.net/wiki/spaces/AC/pages/41910826/Course+Blocks+API+Storage+Cache+Requirements
 """
diff --git a/openedx/core/djangoapps/content/block_structure/config/__init__.py b/openedx/core/djangoapps/content/block_structure/config/__init__.py
index df82bfb068e2..79837df86be6 100644
--- a/openedx/core/djangoapps/content/block_structure/config/__init__.py
+++ b/openedx/core/djangoapps/content/block_structure/config/__init__.py
@@ -9,37 +9,6 @@
 
 from .models import BlockStructureConfiguration
 
-# Switches
-# .. toggle_name: block_structure.storage_backing_for_cache
-# .. toggle_implementation: WaffleSwitch
-# .. toggle_default: False
-# .. toggle_description: When enabled, block structures are stored in a more permanent storage,
-#   like a database, which provides an additional backup for cache misses, instead having them
-#   regenerated. The regenration of block structures is a time consuming process. Therefore,
-#   enabling this switch is recommended for Production.
-# .. toggle_warning: Depends on `BLOCK_STRUCTURES_SETTINGS['STORAGE_CLASS']` and
-#   `BLOCK_STRUCTURES_SETTINGS['STORAGE_KWARGS']`.
-#   This switch will likely be deprecated and removed.
-#   The annotation will be updated with the DEPR ticket once that process has started.
-# .. toggle_use_cases: temporary
-# .. toggle_creation_date: 2017-02-23
-# .. toggle_target_removal_date: 2017-05-23
-# .. toggle_tickets: https://github.com/openedx/edx-platform/pull/14512,
-#   https://github.com/openedx/edx-platform/pull/14770,
-#   https://openedx.atlassian.net/browse/DEPR-145
-STORAGE_BACKING_FOR_CACHE = WaffleSwitch(
-    "block_structure.storage_backing_for_cache", __name__
-)
-
-
-def enable_storage_backing_for_cache_in_request():
-    """
-    Manually override the value of the STORAGE_BACKING_FOR_CACHE switch in the context of the request.
-    This function should not be replicated, as it accesses a protected member, and it shouldn't.
-    """
-    # pylint: disable=protected-access
-    STORAGE_BACKING_FOR_CACHE._cached_switches[STORAGE_BACKING_FOR_CACHE.name] = True
-
 
 @request_cached()
 def num_versions_to_keep():
diff --git a/openedx/core/djangoapps/content/block_structure/management/commands/generate_course_blocks.py b/openedx/core/djangoapps/content/block_structure/management/commands/generate_course_blocks.py
index 4da85f5c52bb..047c945fe406 100644
--- a/openedx/core/djangoapps/content/block_structure/management/commands/generate_course_blocks.py
+++ b/openedx/core/djangoapps/content/block_structure/management/commands/generate_course_blocks.py
@@ -10,7 +10,6 @@
 import openedx.core.djangoapps.content.block_structure.api as api
 import openedx.core.djangoapps.content.block_structure.store as store
 import openedx.core.djangoapps.content.block_structure.tasks as tasks
-from openedx.core.djangoapps.content.block_structure.config import enable_storage_backing_for_cache_in_request
 from openedx.core.lib.command_utils import (
     get_mutually_exclusive_required_option,
     parse_course_keys,
@@ -75,12 +74,6 @@ def add_arguments(self, parser):
             default=0,
             type=int,
         )
-        parser.add_argument(
-            '--with_storage',
-            help='Store the course blocks in Storage, overriding value of the storage_backing_for_cache waffle switch',
-            action='store_true',
-            default=False,
-        )
 
     def handle(self, *args, **options):
 
@@ -129,9 +122,6 @@ def _generate_course_blocks(self, options, course_keys):
         """
         Generates course blocks for the given course_keys per the given options.
         """
-        if options.get('with_storage'):
-            enable_storage_backing_for_cache_in_request()
-
         for course_key in course_keys:
             try:
                 self._generate_for_course(options, course_key)
@@ -150,7 +140,7 @@ def _generate_for_course(self, options, course_key):
             action = tasks.update_course_in_cache_v2 if options.get('force_update') else tasks.get_course_in_cache_v2
             task_options = {'routing_key': options['routing_key']} if options.get('routing_key') else {}
             result = action.apply_async(
-                kwargs=dict(course_id=str(course_key), with_storage=options.get('with_storage')),
+                kwargs=dict(course_id=str(course_key)),
                 **task_options
             )
             log.info('BlockStructure: ENQUEUED generating for course: %s, task_id: %s.', course_key, result.id)
diff --git a/openedx/core/djangoapps/content/block_structure/management/commands/tests/test_generate_course_blocks.py b/openedx/core/djangoapps/content/block_structure/management/commands/tests/test_generate_course_blocks.py
index 5b08c5e95df7..430f374b07cc 100644
--- a/openedx/core/djangoapps/content/block_structure/management/commands/tests/test_generate_course_blocks.py
+++ b/openedx/core/djangoapps/content/block_structure/management/commands/tests/test_generate_course_blocks.py
@@ -85,15 +85,8 @@ def test_all_courses(self, force_update):
             assert mock_update_from_store.call_count == (self.num_courses if force_update else 0)
 
     def test_one_course(self):
-        self._assert_courses_not_in_block_cache(*self.course_keys)
         self.command.handle(courses=[str(self.course_keys[0])])
         self._assert_courses_in_block_cache(self.course_keys[0])
-        self._assert_courses_not_in_block_cache(*self.course_keys[1:])
-        self._assert_courses_not_in_block_storage(*self.course_keys)
-
-    def test_with_storage(self):
-        self.command.handle(with_storage=True, courses=[str(self.course_keys[0])])
-        self._assert_courses_in_block_cache(self.course_keys[0])
         self._assert_courses_in_block_storage(self.course_keys[0])
         self._assert_courses_not_in_block_storage(*self.course_keys[1:])
 
diff --git a/openedx/core/djangoapps/content/block_structure/models.py b/openedx/core/djangoapps/content/block_structure/models.py
index 4872e09346d9..c2837e1a77f0 100644
--- a/openedx/core/djangoapps/content/block_structure/models.py
+++ b/openedx/core/djangoapps/content/block_structure/models.py
@@ -74,7 +74,6 @@ def _bs_model_storage():
     # .. setting_default: None
     # .. setting_description: Specifies the storage used for storage-backed block structure cache.
     #   For more information, check https://github.com/openedx/edx-platform/pull/14571.
-    # .. setting_warnings: Depends on `block_structure.storage_backing_for_cache`.
     storage_class = settings.BLOCK_STRUCTURES_SETTINGS.get('STORAGE_CLASS')
 
     # .. setting_name: BLOCK_STRUCTURES_SETTINGS['STORAGE_KWARGS']
@@ -82,8 +81,7 @@ def _bs_model_storage():
     # .. setting_description: Specifies the keyword arguments needed to setup the storage, which
     #   would be used for storage-backed block structure cache.
     #   For more information, check https://github.com/openedx/edx-platform/pull/14571.
-    # .. setting_warnings: Depends on `BLOCK_STRUCTURES_SETTINGS['STORAGE_CLASS']` and on
-    #   `block_structure.storage_backing_for_cache`.
+    # .. setting_warnings: Depends on `BLOCK_STRUCTURES_SETTINGS['STORAGE_CLASS']`
     storage_kwargs = settings.BLOCK_STRUCTURES_SETTINGS.get('STORAGE_KWARGS', {})
 
     return get_storage(storage_class, **storage_kwargs)
diff --git a/openedx/core/djangoapps/content/block_structure/store.py b/openedx/core/djangoapps/content/block_structure/store.py
index 2342079bdc6e..37a2b57449c0 100644
--- a/openedx/core/djangoapps/content/block_structure/store.py
+++ b/openedx/core/djangoapps/content/block_structure/store.py
@@ -19,26 +19,6 @@
 logger = getLogger(__name__)  # pylint: disable=C0103
 
 
-class StubModel:
-    """
-    Stub model to use when storage backing is disabled.
-    By using this stub, we eliminate the need for extra
-    conditional statements in the code.
-    """
-
-    def __init__(self, root_block_usage_key):
-        self.data_usage_key = root_block_usage_key
-
-    def __str__(self):
-        return str(self.data_usage_key)
-
-    def delete(self):
-        """
-        Noop delete method.
-        """
-        pass  # lint-amnesty, pylint: disable=unnecessary-pass
-
-
 class BlockStructureStore:
     """
     Storage for BlockStructure objects.
@@ -120,13 +100,12 @@ def is_up_to_date(self, root_block_usage_key, modulestore):
         Returns whether the data in storage for the given key is
         already up-to-date with the version in the given modulestore.
         """
-        if config.STORAGE_BACKING_FOR_CACHE.is_enabled():
-            try:
-                bs_model = self._get_model(root_block_usage_key)
-                root_block = modulestore.get_item(root_block_usage_key)
-                return self._version_data_of_model(bs_model) == self._version_data_of_block(root_block)
-            except BlockStructureNotFound:
-                pass
+        try:
+            bs_model = self._get_model(root_block_usage_key)
+            root_block = modulestore.get_item(root_block_usage_key)
+            return self._version_data_of_model(bs_model) == self._version_data_of_block(root_block)
+        except BlockStructureNotFound:
+            pass
 
         return False
 
@@ -134,26 +113,20 @@ def _get_model(self, root_block_usage_key):
         """
         Returns the model associated with the given key.
         """
-        if config.STORAGE_BACKING_FOR_CACHE.is_enabled():
-            return BlockStructureModel.get(root_block_usage_key)
-        else:
-            return StubModel(root_block_usage_key)
+        return BlockStructureModel.get(root_block_usage_key)
 
     def _update_or_create_model(self, block_structure, serialized_data):
         """
         Updates or creates the model for the given block_structure
         and serialized_data.
         """
-        if config.STORAGE_BACKING_FOR_CACHE.is_enabled():
-            root_block = block_structure[block_structure.root_block_usage_key]
-            bs_model, _ = BlockStructureModel.update_or_create(
-                serialized_data,
-                data_usage_key=block_structure.root_block_usage_key,
-                **self._version_data_of_block(root_block)
-            )
-            return bs_model
-        else:
-            return StubModel(block_structure.root_block_usage_key)
+        root_block = block_structure[block_structure.root_block_usage_key]
+        bs_model, _ = BlockStructureModel.update_or_create(
+            serialized_data,
+            data_usage_key=block_structure.root_block_usage_key,
+            **self._version_data_of_block(root_block)
+        )
+        return bs_model
 
     def _add_to_cache(self, serialized_data, bs_model):
         """
@@ -186,9 +159,6 @@ def _get_from_store(self, bs_model):
         Raises:
              BlockStructureNotFound if not found.
         """
-        if not config.STORAGE_BACKING_FOR_CACHE.is_enabled():
-            raise BlockStructureNotFound(bs_model.data_usage_key)
-
         return bs_model.get_serialized_data()
 
     def _serialize(self, block_structure):
@@ -226,14 +196,9 @@ def _deserialize(self, serialized_data, root_block_usage_key):
     def _encode_root_cache_key(bs_model):
         """
         Returns the cache key to use for the given
-        BlockStructureModel or StubModel.
+        BlockStructureModel.
         """
-        if config.STORAGE_BACKING_FOR_CACHE.is_enabled():
-            return str(bs_model)
-        return "v{version}.root.key.{root_usage_key}".format(
-            version=str(BlockStructureBlockData.VERSION),
-            root_usage_key=str(bs_model.data_usage_key),
-        )
+        return str(bs_model)
 
     @staticmethod
     def _version_data_of_block(root_block):
diff --git a/openedx/core/djangoapps/content/block_structure/tasks.py b/openedx/core/djangoapps/content/block_structure/tasks.py
index 7c3c2805fb10..5796b8167b2b 100644
--- a/openedx/core/djangoapps/content/block_structure/tasks.py
+++ b/openedx/core/djangoapps/content/block_structure/tasks.py
@@ -14,7 +14,6 @@
 
 from xmodule.capa.responsetypes import LoncapaProblemError
 from openedx.core.djangoapps.content.block_structure import api
-from openedx.core.djangoapps.content.block_structure.config import enable_storage_backing_for_cache_in_request
 from xmodule.modulestore.exceptions import ItemNotFoundError  # lint-amnesty, pylint: disable=wrong-import-order
 
 log = logging.getLogger('edx.celery.task')
@@ -43,8 +42,6 @@ def update_course_in_cache_v2(self, **kwargs):
     Updates the course blocks (mongo -> BlockStructure) for the specified course.
     Keyword Arguments:
         course_id (string) - The string serialized value of the course key.
-        with_storage (boolean) - Whether or not storage backing should be
-            enabled for the generated block structure(s).
     """
     _update_course_in_cache(self, **kwargs)
 
@@ -62,8 +59,6 @@ def _update_course_in_cache(self, **kwargs):
     """
     Updates the course blocks (mongo -> BlockStructure) for the specified course.
     """
-    if kwargs.get('with_storage'):
-        enable_storage_backing_for_cache_in_request()
     _call_and_retry_if_needed(self, api.update_course_in_cache, **kwargs)
 
 
@@ -74,8 +69,6 @@ def get_course_in_cache_v2(self, **kwargs):
     Gets the course blocks for the specified course, updating the cache if needed.
     Keyword Arguments:
         course_id (string) - The string serialized value of the course key.
-        with_storage (boolean) - Whether or not storage backing should be
-            enabled for any generated block structure(s).
     """
     _get_course_in_cache(self, **kwargs)
 
@@ -93,8 +86,6 @@ def _get_course_in_cache(self, **kwargs):
     """
     Gets the course blocks for the specified course, updating the cache if needed.
     """
-    if kwargs.get('with_storage'):
-        enable_storage_backing_for_cache_in_request()
     _call_and_retry_if_needed(self, api.get_course_in_cache, **kwargs)
 
 
diff --git a/openedx/core/djangoapps/content/block_structure/tests/helpers.py b/openedx/core/djangoapps/content/block_structure/tests/helpers.py
index 93655c79e16e..f0a20554e6d2 100644
--- a/openedx/core/djangoapps/content/block_structure/tests/helpers.py
+++ b/openedx/core/djangoapps/content/block_structure/tests/helpers.py
@@ -274,10 +274,14 @@ def create_block_structure(self, children_map, block_structure_cls=BlockStructur
         # create empty block structure
         block_structure = block_structure_cls(root_block_usage_key=self.block_key_factory(0))
 
-        # _add_relation
+        # _add_relation and blocks
         for parent, children in enumerate(children_map):
+            if isinstance(block_structure, BlockStructureBlockData):
+                block_structure._get_or_create_block(self.block_key_factory(parent))  # pylint: disable=protected-access
             for child in children:
                 block_structure._add_relation(self.block_key_factory(parent), self.block_key_factory(child))  # pylint: disable=protected-access
+                if isinstance(block_structure, BlockStructureBlockData):
+                    block_structure._get_or_create_block(self.block_key_factory(child))  # pylint: disable=protected-access
         return block_structure
 
     def get_parents_map(self, children_map):
diff --git a/openedx/core/djangoapps/content/block_structure/tests/test_factory.py b/openedx/core/djangoapps/content/block_structure/tests/test_factory.py
index b9fb8c5e1077..00efa393d8fa 100644
--- a/openedx/core/djangoapps/content/block_structure/tests/test_factory.py
+++ b/openedx/core/djangoapps/content/block_structure/tests/test_factory.py
@@ -5,6 +5,7 @@
 import pytest
 from django.test import TestCase
 
+from opaque_keys.edx.keys import CourseKey
 from xmodule.modulestore.exceptions import ItemNotFoundError
 
 from ..exceptions import BlockStructureNotFound
@@ -18,14 +19,22 @@ class TestBlockStructureFactory(TestCase, ChildrenMapTestMixin):
     Tests for BlockStructureFactory
     """
 
+    def block_key_factory(self, block_id):
+        """
+        Returns a usage_key object for the given block_id.
+        This overrides the method in the ChildrenMapTestMixin.
+        """
+        return CourseKey.from_string("course-v1:org+course+run").make_usage_key("html", str(block_id))
+
     def setUp(self):
         super().setUp()
         self.children_map = self.SIMPLE_CHILDREN_MAP
         self.modulestore = MockModulestoreFactory.create(self.children_map, self.block_key_factory)
 
     def test_from_modulestore(self):
+        usage_key = CourseKey.from_string("course-v1:org+course+run").make_usage_key("html", "0")
         block_structure = BlockStructureFactory.create_from_modulestore(
-            root_block_usage_key=0, modulestore=self.modulestore
+            root_block_usage_key=usage_key, modulestore=self.modulestore
         )
         self.assert_block_structure(block_structure, self.children_map)
 
@@ -48,15 +57,18 @@ def test_from_cache(self):
 
     def test_from_cache_none(self):
         store = BlockStructureStore(MockCache())
+        # Non-existent usage key
+        usage_key = CourseKey.from_string("course-v1:org+course+run").make_usage_key("html", "0")
         with pytest.raises(BlockStructureNotFound):
             BlockStructureFactory.create_from_store(
-                root_block_usage_key=0,
+                root_block_usage_key=usage_key,
                 block_structure_store=store,
             )
 
     def test_new(self):
+        usage_key = CourseKey.from_string("course-v1:org+course+run").make_usage_key("html", "0")
         block_structure = BlockStructureFactory.create_from_modulestore(
-            root_block_usage_key=0, modulestore=self.modulestore
+            root_block_usage_key=usage_key, modulestore=self.modulestore
         )
         new_structure = BlockStructureFactory.create_new(
             block_structure.root_block_usage_key,
diff --git a/openedx/core/djangoapps/content/block_structure/tests/test_manager.py b/openedx/core/djangoapps/content/block_structure/tests/test_manager.py
index 8e5b585ca879..493e4c34714f 100644
--- a/openedx/core/djangoapps/content/block_structure/tests/test_manager.py
+++ b/openedx/core/djangoapps/content/block_structure/tests/test_manager.py
@@ -5,10 +5,8 @@
 import pytest
 import ddt
 from django.test import TestCase
-from edx_toggles.toggles.testutils import override_waffle_switch
 
 from ..block_structure import BlockStructureBlockData
-from ..config import STORAGE_BACKING_FOR_CACHE
 from ..exceptions import UsageKeyNotInBlockStructure
 from ..manager import BlockStructureManager
 from ..transformers import BlockStructureTransformers
@@ -177,20 +175,18 @@ def test_get_collected_cached(self):
         self.collect_and_verify(expect_modulestore_called=False, expect_cache_updated=False)
         assert TestTransformer1.collect_call_count == 1
 
-    @ddt.data(True, False)
-    def test_update_collected_if_needed(self, with_storage_backing):
-        with override_waffle_switch(STORAGE_BACKING_FOR_CACHE, active=with_storage_backing):
-            with mock_registered_transformers(self.registered_transformers):
-                assert TestTransformer1.collect_call_count == 0
+    def test_update_collected_if_needed(self):
+        with mock_registered_transformers(self.registered_transformers):
+            assert TestTransformer1.collect_call_count == 0
 
-                self.bs_manager.update_collected_if_needed()
-                assert TestTransformer1.collect_call_count == 1
+            self.bs_manager.update_collected_if_needed()
+            assert TestTransformer1.collect_call_count == 1
 
-                self.bs_manager.update_collected_if_needed()
-                expected_count = 1 if with_storage_backing else 2
-                assert TestTransformer1.collect_call_count == expected_count
+            self.bs_manager.update_collected_if_needed()
+            expected_count = 1
+            assert TestTransformer1.collect_call_count == expected_count
 
-                self.collect_and_verify(expect_modulestore_called=False, expect_cache_updated=False)
+            self.collect_and_verify(expect_modulestore_called=False, expect_cache_updated=False)
 
     def test_get_collected_transformer_version(self):
         self.collect_and_verify(expect_modulestore_called=True, expect_cache_updated=True)
@@ -212,8 +208,8 @@ def test_get_collected_transformer_version(self):
     def test_get_collected_structure_version(self):
         self.collect_and_verify(expect_modulestore_called=True, expect_cache_updated=True)
         BlockStructureBlockData.VERSION += 1
-        self.collect_and_verify(expect_modulestore_called=True, expect_cache_updated=True)
-        assert TestTransformer1.collect_call_count == 2
+        self.collect_and_verify(expect_modulestore_called=False, expect_cache_updated=False)
+        assert TestTransformer1.collect_call_count == 1
 
     def test_clear(self):
         self.collect_and_verify(expect_modulestore_called=True, expect_cache_updated=True)
diff --git a/openedx/core/djangoapps/content/block_structure/tests/test_store.py b/openedx/core/djangoapps/content/block_structure/tests/test_store.py
index 8d6fc8017d0f..63a02efa7af8 100644
--- a/openedx/core/djangoapps/content/block_structure/tests/test_store.py
+++ b/openedx/core/djangoapps/content/block_structure/tests/test_store.py
@@ -4,11 +4,9 @@
 
 import pytest
 import ddt
-from edx_toggles.toggles.testutils import override_waffle_switch
 
 from openedx.core.djangolib.testing.utils import CacheIsolationTestCase
 
-from ..config import STORAGE_BACKING_FOR_CACHE
 from ..config.models import BlockStructureConfiguration
 from ..exceptions import BlockStructureNotFound
 from ..store import BlockStructureStore
@@ -46,40 +44,27 @@ def add_transformers(self):
                 value=f'{transformer.name()} val',
             )
 
-    @ddt.data(True, False)
-    def test_get_none(self, with_storage_backing):
-        with override_waffle_switch(STORAGE_BACKING_FOR_CACHE, active=with_storage_backing):
-            with pytest.raises(BlockStructureNotFound):
-                self.store.get(self.block_structure.root_block_usage_key)
-
-    @ddt.data(True, False)
-    def test_add_and_get(self, with_storage_backing):
-        with override_waffle_switch(STORAGE_BACKING_FOR_CACHE, active=with_storage_backing):
-            self.store.add(self.block_structure)
-            stored_value = self.store.get(self.block_structure.root_block_usage_key)
-            assert stored_value is not None
-            self.assert_block_structure(stored_value, self.children_map)
+    def test_get_none(self):
+        with pytest.raises(BlockStructureNotFound):
+            self.store.get(self.block_structure.root_block_usage_key)
 
-    @ddt.data(True, False)
-    def test_delete(self, with_storage_backing):
-        with override_waffle_switch(STORAGE_BACKING_FOR_CACHE, active=with_storage_backing):
-            self.store.add(self.block_structure)
-            self.store.delete(self.block_structure.root_block_usage_key)
-            with pytest.raises(BlockStructureNotFound):
-                self.store.get(self.block_structure.root_block_usage_key)
+    def test_add_and_get(self):
+        self.store.add(self.block_structure)
+        stored_value = self.store.get(self.block_structure.root_block_usage_key)
+        assert stored_value is not None
+        self.assert_block_structure(stored_value, self.children_map)
 
-    def test_uncached_without_storage(self):
+    def test_delete(self):
         self.store.add(self.block_structure)
-        self.mock_cache.map.clear()
+        self.store.delete(self.block_structure.root_block_usage_key)
         with pytest.raises(BlockStructureNotFound):
             self.store.get(self.block_structure.root_block_usage_key)
 
     def test_uncached_with_storage(self):
-        with override_waffle_switch(STORAGE_BACKING_FOR_CACHE, active=True):
-            self.store.add(self.block_structure)
-            self.mock_cache.map.clear()
-            stored_value = self.store.get(self.block_structure.root_block_usage_key)
-            self.assert_block_structure(stored_value, self.children_map)
+        self.store.add(self.block_structure)
+        self.mock_cache.map.clear()
+        stored_value = self.store.get(self.block_structure.root_block_usage_key)
+        self.assert_block_structure(stored_value, self.children_map)
 
     @ddt.data(1, 5, None)
     def test_cache_timeout(self, timeout):