Disable retry logic in on-demand content streaming

Our RemoteArtifact streaming uses chunked transfer, so once it has
started a response, we can't fix anything anymore in case an error.

Because of that, the retry logic (any of the timeout, digest, etc) on
the http downloader is just delaying a response that can't be possible
right anymore.

Closes pulp#5947

CHANGES/5937.bugfix

@@ -0,0 +1,2 @@
+Disable retry logic on the context of content-app on-demand streaming, as we can't recover
+from any errors after starting the streaming process (chunked transfer).

pulpcore/content/handler.py

@@ -1142,7 +1142,7 @@ async def finalize():
         original_finalize = downloader.finalize
         downloader.finalize = finalize
         try:
-            download_result = await downloader.run()
+            download_result = await downloader.run(retry=False)
         except DigestValidationError:
             remote_artifact.failed_at = timezone.now()
             await remote_artifact.asave()

pulpcore/download/http.py

@@ -212,9 +212,9 @@ async def _handle_response(self, response):
             headers=response.headers,
         )
 
-    async def run(self, extra_data=None):
+    async def run(self, extra_data=None, retry=True):
         """
-        Run the downloader with concurrency restriction and retry logic.
+        Run the downloader with concurrency restriction and optional retry logic.
 
         This method acquires `self.semaphore` before calling the actual download implementation
         contained in `_run()`. This ensures that the semaphore stays acquired even as the `backoff`
@@ -240,14 +240,22 @@ async def run(self, extra_data=None):
             SizeValidationError,
         )
 
-        async with self.semaphore:
+        def noop_decorator(func):
+            return func
 
-            @backoff.on_exception(
+        if retry is True:
+            backoff_decorator = backoff.on_exception(
                 backoff.expo,
                 retryable_errors,
                 max_tries=self.max_retries + 1,
                 giveup=http_giveup_handler,
             )
+        else:
+            backoff_decorator = noop_decorator
+
+        async with self.semaphore:
+
+            @backoff_decorator
             async def download_wrapper():
                 self._ensure_no_broken_file()
                 try: