diff --git a/openedx/core/djangoapps/content_libraries/tasks.py b/openedx/core/djangoapps/content_libraries/tasks.py
index 55a2329681e0..fef25f039f46 100644
--- a/openedx/core/djangoapps/content_libraries/tasks.py
+++ b/openedx/core/djangoapps/content_libraries/tasks.py
@@ -32,7 +32,11 @@
 from opaque_keys.edx.locator import BlockUsageLocator
 from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
 from openedx.core.lib import ensure_cms
-from openedx_learning.api.authoring import get_entity_links, get_or_create_learning_context_link_status
+from openedx_learning.api.authoring import (
+    get_entity_links,
+    get_or_create_learning_context_link_status,
+    update_learning_context_link_status,
+)
 from openedx_learning.api.authoring_models import LearningContextLinksStatusChoices
 from xmodule.capa_block import ProblemBlock
 from xmodule.library_content_block import ANY_CAPA_TYPE_VALUE, LegacyLibraryContentBlock
@@ -210,18 +214,27 @@ def create_or_update_upstream_links(course_key_str: str, force: bool = False, cr
         return
     store = modulestore()
     course_key = CourseKey.from_string(course_key_str)
-    course_status.status = LearningContextLinksStatusChoices.PROCESSING
-    course_status.save()
+    update_learning_context_link_status(
+        context_key=course_key_str,
+        status=LearningContextLinksStatusChoices.PROCESSING,
+        updated=created,
+    )
     try:
         course_name = CourseOverview.get_from_id(course_key).display_name_with_default
     except CourseOverview.DoesNotExist:
         TASK_LOGGER.exception(f'Could not find course: {course_key_str}')
+        update_learning_context_link_status(
+            context_key=course_key_str,
+            status=LearningContextLinksStatusChoices.FAILED,
+        )
         return
     xblocks = store.get_items(course_key, settings={"upstream": lambda x: x is not None})
     for xblock in xblocks:
         api.create_or_update_xblock_upstream_link(xblock, course_key_str, course_name, created)
-    course_status.status = LearningContextLinksStatusChoices.COMPLETED
-    course_status.save()
+    update_learning_context_link_status(
+        context_key=course_key_str,
+        status=LearningContextLinksStatusChoices.COMPLETED,
+    )
 
 
 @shared_task(base=LoggedTask)
diff --git a/openedx/core/djangoapps/content_libraries/tests/test_upstream_downstream_links.py b/openedx/core/djangoapps/content_libraries/tests/test_upstream_downstream_links.py
index 7215c8576c8d..96a9dc311b33 100644
--- a/openedx/core/djangoapps/content_libraries/tests/test_upstream_downstream_links.py
+++ b/openedx/core/djangoapps/content_libraries/tests/test_upstream_downstream_links.py
@@ -10,6 +10,8 @@
 from django.utils import timezone
 from freezegun import freeze_time
 
+from openedx_learning.api.authoring_models import LearningContextLinksStatus, LearningContextLinksStatusChoices
+
 from openedx.core.djangolib.testing.utils import skip_unless_cms
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import BlockFactory, CourseFactory
@@ -137,12 +139,18 @@ def test_create_or_update_upstream_links_task(self, mock_api):
         """
         Test task create_or_update_upstream_links for a course
         """
+        assert not LearningContextLinksStatus.objects.filter(context_key=str(self.course_key)).exists()
         create_or_update_upstream_links(str(self.course_key), force=False)
         expected_calls = [
             (self.component_1.usage_key, str(self.course_key), self.course.display_name_with_default, self.now),
             (self.component_2.usage_key, str(self.course_key), self.course.display_name_with_default, self.now),
         ]
         assert [(x[0][0].usage_key, x[0][1], x[0][2], x[0][3]) for x in mock_api.call_args_list] == expected_calls
+        assert LearningContextLinksStatus.objects.filter(context_key=str(self.course_key)).exists()
+        assert LearningContextLinksStatus.objects.filter(
+            context_key=str(self.course_key)
+        ).first().status == LearningContextLinksStatusChoices.COMPLETED
+
         mock_api.reset_mock()
         # call again with same course, it should not be processed again
         # as its LearningContextLinksStatusChoices = COMPLETED
@@ -168,3 +176,19 @@ def test_create_or_update_xblock_upstream_link(self, mock_api):
         # call for xblock with no upstream
         create_or_update_xblock_upstream_link(str(self.component_3.usage_key))
         mock_api.assert_not_called()
+
+    @patch(
+        'openedx.core.djangoapps.content_libraries.api.create_or_update_xblock_upstream_link'
+    )
+    def test_create_or_update_upstream_links_task_for_invalid_course(self, mock_api):
+        """
+        Test task create_or_update_upstream_links for an invalid course key.
+        """
+        course_key = "course-v1:non+existent+course"
+        assert not LearningContextLinksStatus.objects.filter(context_key=course_key).exists()
+        create_or_update_upstream_links(course_key, force=False)
+        mock_api.assert_not_called()
+        assert LearningContextLinksStatus.objects.filter(context_key=course_key).exists()
+        assert LearningContextLinksStatus.objects.filter(
+            context_key=course_key
+        ).first().status == LearningContextLinksStatusChoices.FAILED