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 #5937

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

@@ -1138,7 +1138,9 @@ async def finalize():
         original_finalize = downloader.finalize
         downloader.finalize = finalize
         try:
-            download_result = await downloader.run()
+            download_result = await downloader.run(
+                extra_data={"disable_retry_list": (DigestValidationError,)}
+            )
         except DigestValidationError:
             await downloader.session.close()
             close_tcp_connection(request.transport._sock)

pulpcore/download/http.py

@@ -214,20 +214,22 @@ async def _handle_response(self, response):
 
     async def run(self, extra_data=None):
         """
-        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`
         wrapper around `_run()`, handles backoff-and-retry logic.
 
         Args:
-            extra_data (dict): Extra data passed to the downloader.
+            extra_data (dict): Extra data passed to the downloader:
+                disable_retry_list: List of exceptions which should not be retried.
 
         Returns:
             [pulpcore.plugin.download.DownloadResult][] from `_run()`.
 
         """
-        retryable_errors = (
+        disable_retry_list = [] if not extra_data else extra_data.get("disable_retry_list", [])
+        default_retryable_errors = (
             aiohttp.ClientConnectorSSLError,
             aiohttp.ClientConnectorError,
             aiohttp.ClientOSError,
@@ -240,6 +242,9 @@ async def run(self, extra_data=None):
             SizeValidationError,
         )
 
+        retryable_errors = tuple(
+            [e for e in default_retryable_errors if e not in disable_retry_list]
+        )
         async with self.semaphore:
 
             @backoff.on_exception(