Update remote artifact urls on sync if the remote or repo changes

closes: #9395
https://pulp.plan.io/issues/9395

CHANGES/9395.bugfix

@@ -0,0 +1 @@
+Fixed an issue where on_demand content might not be downloaded properly if the remote URL was changed (even if re-synced).

pulpcore/plugin/stages/artifact_stages.py

@@ -276,13 +276,11 @@ async def run(self):
             The coroutine for this stage.
         """
         async for batch in self.batches():
-            await sync_to_async(RemoteArtifact.objects.bulk_get_or_create)(
-                await self._needed_remote_artifacts(batch)
-            )
+            await self._handle_remote_artifacts(batch)
             for d_content in batch:
                 await self.put(d_content)
 
-    async def _needed_remote_artifacts(self, batch):
+    async def _handle_remote_artifacts(self, batch):
         """
         Build a list of only :class:`~pulpcore.plugin.models.RemoteArtifact` that need
         to be created for the batch.
@@ -318,7 +316,8 @@ async def _needed_remote_artifacts(self, batch):
         #
         # We can end up with duplicates (diff pks, same sha256) in the sequence below,
         # so we store by-sha256 and then return the final values
-        needed_ras = {}  # { str(<sha256>): RemoteArtifact, ... }
+        ras_to_create = {}  # { str(<sha256>): RemoteArtifact, ... }
+        ras_to_update = {}
         for d_content in batch:
             for d_artifact in d_content.d_artifacts:
                 if not d_artifact.remote:
@@ -379,14 +378,25 @@ async def _needed_remote_artifacts(self, batch):
                 async for remote_artifact in sync_to_async_iterable(
                     content_artifact._remote_artifact_saver_ras
                 ):
-                    if remote_artifact.remote_id == d_artifact.remote.pk:
+                    if d_artifact.remote.url == remote_artifact.url:
+                        break
+
+                    if d_artifact.remote.pk == remote_artifact.remote_id:
+                        key = f"{str(content_artifact.pk)}-{str(d_artifact.remote.pk)}"
+                        remote_artifact.url = d_artifact.url
+                        ras_to_update[key] = remote_artifact
                         break
                 else:
                     remote_artifact = self._create_remote_artifact(d_artifact, content_artifact)
                     key = f"{str(content_artifact.pk)}-{str(d_artifact.remote.pk)}"
-                    needed_ras[key] = remote_artifact
+                    ras_to_create[key] = remote_artifact
 
-        return list(needed_ras.values())
+        if ras_to_create:
+            await sync_to_async(RemoteArtifact.objects.bulk_create)(list(ras_to_create.values()))
+        if ras_to_update:
+            await sync_to_async(RemoteArtifact.objects.bulk_update)(
+                list(ras_to_update.values()), fields=["url"]
+            )
 
     @staticmethod
     def _create_remote_artifact(d_artifact, content_artifact):