Catch urllib3 ProtocolError on streamed responses

.changes/next-release/bugfix-s3-27431.json

@@ -0,0 +1,5 @@
+{
+  "type": "bugfix",
+  "category": "``s3``",
+  "description": "Wrap ``urllib3.exceptions.ProtocolError`` raised while streaming a response body in ``ResponseStreamingError`` so that part-level download retries apply when a connection is reset mid-transfer."
+}

awscli/botocore/exceptions.py

@@ -146,6 +146,10 @@ class ProxyConnectionError(ConnectionError):
     fmt = 'Failed to connect to proxy URL: "{proxy_url}"'
 
 
+class ResponseStreamingError(HTTPClientError):
+    fmt = 'An error occurred while reading from response stream: {error}'
+
+
 class NoCredentialsError(BotoCoreError):
     """
     No credentials could be found.

awscli/botocore/response.py

@@ -24,8 +24,13 @@
     json,  # noqa
     set_socket_timeout,
 )
-from botocore.exceptions import IncompleteReadError, ReadTimeoutError
+from botocore.exceptions import (
+    IncompleteReadError,
+    ReadTimeoutError,
+    ResponseStreamingError,
+)
 from botocore.hooks import first_non_none_response  # noqa
+from urllib3.exceptions import ProtocolError as URLLib3ProtocolError
 from urllib3.exceptions import ReadTimeoutError as URLLib3ReadTimeoutError
 
 logger = logging.getLogger(__name__)
@@ -87,6 +92,8 @@ def read(self, amt=None):
         except URLLib3ReadTimeoutError as e:
             # TODO: the url will be None as urllib3 isn't setting it yet
             raise ReadTimeoutError(endpoint_url=e.url, error=e)
+        except URLLib3ProtocolError as e:
+            raise ResponseStreamingError(error=e)
         self._amount_read += len(chunk)
         if amt is None or (not chunk and amt > 0):
             # If the server sends empty contents or

awscli/s3transfer/utils.py

@@ -21,7 +21,11 @@
 import threading
 from collections import defaultdict
 
-from botocore.exceptions import IncompleteReadError, ReadTimeoutError
+from botocore.exceptions import (
+    IncompleteReadError,
+    ReadTimeoutError,
+    ResponseStreamingError,
+)
 from botocore.httpchecksum import DEFAULT_CHECKSUM_ALGORITHM, AwsChunkedWrapper
 from botocore.utils import is_s3express_bucket
 from s3transfer.compat import SOCKET_ERROR, fallocate, rename_file
@@ -41,6 +45,7 @@
     SOCKET_ERROR,
     ReadTimeoutError,
     IncompleteReadError,
+    ResponseStreamingError,
 )
 
 

tests/unit/botocore/test_response.py

@@ -16,8 +16,13 @@
 import botocore
 from botocore import response
 from botocore.awsrequest import AWSRequest, AWSResponse
-from botocore.exceptions import IncompleteReadError, ReadTimeoutError
+from botocore.exceptions import (
+    IncompleteReadError,
+    ReadTimeoutError,
+    ResponseStreamingError,
+)
 from dateutil.tz import tzutc
+from urllib3.exceptions import ProtocolError as URLLib3ProtocolError
 from urllib3.exceptions import ReadTimeoutError as URLLib3ReadTimeoutError
 
 from tests import unittest
@@ -169,6 +174,20 @@ def geturl(*args, **kwargs):
         with self.assertRaises(ReadTimeoutError):
             stream.read()
 
+    def test_catches_urllib3_protocol_error(self):
+        class ProtocolErrorBody:
+            def read(*args, **kwargs):
+                raise URLLib3ProtocolError(None, None, None)
+
+            def geturl(*args, **kwargs):
+                return 'http://example.com'
+
+        stream = response.StreamingBody(
+            ProtocolErrorBody(), content_length=None
+        )
+        with self.assertRaises(ResponseStreamingError):
+            stream.read()
+
     def test_streaming_line_abstruse_newline_standard(self):
         for chunk_size in range(1, 30):
             body = io.BytesIO(b'1234567890\r\n1234567890\r\n12345\r\n')

tests/unit/s3transfer/test_download.py

@@ -19,6 +19,7 @@
 
 from awscrt import checksums as crt_checksums
 from botocore.config import Config
+from botocore.response import StreamingBody
 from s3transfer.bandwidth import BandwidthLimiter
 from s3transfer.checksums import (
     FullObjectChecksum,
@@ -45,6 +46,7 @@
 from s3transfer.exceptions import RetriesExceededError
 from s3transfer.futures import IN_MEMORY_DOWNLOAD_TAG, BoundedExecutor
 from s3transfer.utils import CallArgs, OSUtils
+from urllib3.exceptions import ProtocolError as URLLib3ProtocolError
 
 from tests import (
     BaseSubmissionTaskTest,
@@ -767,6 +769,31 @@ def test_retries_succeeds(self):
         self.stubber.assert_no_pending_responses()
         self.assert_io_writes([(0, self.content)])
 
+    def test_retries_urllib3_protocol_error(self):
+        # A ProtocolError raised while streaming the response body (e.g. the
+        # connection is reset mid-download) is wrapped by StreamingBody into a
+        # ResponseStreamingError, which is retryable.
+        self.stubber.add_response(
+            'get_object',
+            service_response={
+                'Body': StreamingBody(
+                    StreamWithError(self.stream, URLLib3ProtocolError),
+                    content_length=None,
+                )
+            },
+            expected_params={'Bucket': self.bucket, 'Key': self.key},
+        )
+        self.stubber.add_response(
+            'get_object',
+            service_response={'Body': BytesIO(self.content)},
+            expected_params={'Bucket': self.bucket, 'Key': self.key},
+        )
+        task = self.get_download_task()
+        task()
+
+        self.stubber.assert_no_pending_responses()
+        self.assert_io_writes([(0, self.content)])
+
     def test_retries_failure(self):
         for _ in range(self.max_attempts):
             self.stubber.add_response(