Properly update present ContentAtifacts after immediate sync

fixes: #9101

CHANGES/9101.bugfix

@@ -0,0 +1 @@
+Artifacts are now being properly updated for Content after switching from 'on_demand' to 'immediate'.

pulpcore/plugin/stages/content_stages.py

@@ -98,6 +98,9 @@ async def run(self):
         """
         async for batch in self.batches():
             content_artifact_bulk = []
+            to_update_ca_query = ContentArtifact.objects.none()
+            to_update_ca_bulk = []
+            to_update_ca_artifact = {}
             with transaction.atomic():
                 await self._pre_save(batch)
 
@@ -115,19 +118,37 @@ async def run(self):
                                 )
                             except ObjectDoesNotExist:
                                 raise e
+                        else:
+                            for d_artifact in d_content.d_artifacts:
+                                if not d_artifact.artifact._state.adding:
+                                    artifact = d_artifact.artifact
+                                else:
+                                    # set to None for on-demand synced artifacts
+                                    artifact = None
+                                content_artifact = ContentArtifact(
+                                    content=d_content.content,
+                                    artifact=artifact,
+                                    relative_path=d_artifact.relative_path,
+                                )
+                                content_artifact_bulk.append(content_artifact)
                             continue
-                        for d_artifact in d_content.d_artifacts:
-                            if not d_artifact.artifact._state.adding:
-                                artifact = d_artifact.artifact
-                            else:
-                                # set to None for on-demand synced artifacts
-                                artifact = None
-                            content_artifact = ContentArtifact(
-                                content=d_content.content,
-                                artifact=artifact,
-                                relative_path=d_artifact.relative_path,
+                    # When the Content already exists, check if ContentArtifacts need to be updated
+                    for d_artifact in d_content.d_artifacts:
+                        if not d_artifact.artifact._state.adding:
+                            # the artifact is already present in the database; update references
+                            # Creating one large query and one large dictionary
+                            to_update_ca_query |= ContentArtifact.objects.filter(
+                                content=d_content.content, relative_path=d_artifact.relative_path
                             )
-                            content_artifact_bulk.append(content_artifact)
+                            key = (d_content.content.pk, d_artifact.relative_path)
+                            to_update_ca_artifact[key] = d_artifact.artifact
+                # Query db once and update each object in memory for bulk_update call
+                for content_artifact in to_update_ca_query.iterator():
+                    key = (content_artifact.content_id, content_artifact.relative_path)
+                    # Maybe remove dict elements after to reduce memory?
+                    content_artifact.artifact = to_update_ca_artifact[key]
+                    to_update_ca_bulk.append(content_artifact)
+                ContentArtifact.objects.bulk_update(to_update_ca_bulk, ["artifact"])
                 ContentArtifact.objects.bulk_get_or_create(content_artifact_bulk)
                 await self._post_save(batch)
             for declarative_content in batch:

pulpcore/tests/functional/api/using_plugin/test_sync.py

@@ -0,0 +1,73 @@
+from pulp_smash.pulp3.bindings import delete_orphans, monitor_task, PulpTestCase
+from pulp_smash.pulp3.utils import gen_repo
+
+from pulpcore.tests.functional.api.using_plugin.utils import (
+    gen_file_client,
+    gen_file_remote,
+)
+from pulpcore.client.pulp_file import (
+    ContentFilesApi,
+    RepositorySyncURL,
+    RepositoriesFileApi,
+    RemotesFileApi,
+)
+from pulpcore.tests.functional.api.using_plugin.utils import (  # noqa:F401
+    set_up_module as setUpModule,
+)
+
+
+class MultiplePolicySyncTestCase(PulpTestCase):
+    """
+    This test ensures that content artifacts are properly updated when syncing multiple
+    times with different policies, specifically from 'on_demand' to 'immediate'
+
+    This test targets the following issue:
+    * `Pulp #9101 <https://pulp.plan.io/issues/9101>`_
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        """Clean out Pulp before testing."""
+        delete_orphans()
+        client = gen_file_client()
+        cls.cont_api = ContentFilesApi(client)
+        cls.repo_api = RepositoriesFileApi(client)
+        cls.remote_api = RemotesFileApi(client)
+
+    def tearDown(self):
+        """Clean up Pulp after testing."""
+        self.doCleanups()
+        delete_orphans()
+
+    def test_ondemand_to_immediate_sync(self):
+        """Checks that content artifacts are updated following on-demand -> immediate sync."""
+        # Ensure that no content is present
+        content_response = self.cont_api.list(limit=1)
+        if content_response.count > 0:
+            self.skipTest("Please remove all file content before running this test")
+
+        # Create and sync repo w/ on_demand policy
+        repo = self.repo_api.create(gen_repo())
+        remote = self.remote_api.create(gen_file_remote(policy="on_demand"))
+        body = RepositorySyncURL(remote=remote.pulp_href)
+        monitor_task(self.repo_api.sync(repo.pulp_href, body).task)
+        self.addCleanup(self.repo_api.delete, repo.pulp_href)
+        self.addCleanup(self.remote_api.delete, remote.pulp_href)
+
+        # Check content is present, but no artifacts are there
+        content_response = self.cont_api.list()
+        self.assertEqual(content_response.count, 3)
+        for content in content_response.results:
+            self.assertEqual(content.artifact, None)
+
+        # Sync again w/ immediate policy
+        remote = self.remote_api.create(gen_file_remote())
+        body = RepositorySyncURL(remote=remote.pulp_href)
+        monitor_task(self.repo_api.sync(repo.pulp_href, body).task)
+        self.addCleanup(self.remote_api.delete, remote.pulp_href)
+
+        # Check content is still present, but artifacts are now there
+        content_response = self.cont_api.list()
+        self.assertEqual(content_response.count, 3)
+        for content in content_response.results:
+            self.assertNotEqual(content.artifact, None)