diff --git a/tests/providers/s3compatsigv4/test_provider.py b/tests/providers/s3compatsigv4/test_provider.py
index bd21e0752..ef7b3f284 100644
--- a/tests/providers/s3compatsigv4/test_provider.py
+++ b/tests/providers/s3compatsigv4/test_provider.py
@@ -8,6 +8,7 @@
 import datetime
 import aiohttpretty
 from http import client
+from http import HTTPStatus
 from urllib import parse
 from unittest import mock
 
@@ -1188,15 +1189,191 @@ async def test_chunked_upload_aborted_success(self, provider, upload_parts_heade
 
         with pytest.raises(exceptions.UploadError) as exc:
             await provider._chunked_upload(file_stream, path)
+        # The abort has SUCCEEDED (return value True), so the "manual clean-up"
+        # warning must NOT be appended to the error message.
         msg = 'An unexpected error has occurred during the multi-part upload.'
-        msg += '  The abort action failed to clean up the temporary file parts generated ' \
-               'during the upload process.  Please manually remove them.'
         assert str(exc.value) == ', '.join(['500', msg])
 
         provider._create_upload_session.assert_called_with(path)
         provider._upload_parts.assert_called_with(file_stream, path, upload_id)
         provider._abort_chunked_upload.assert_called_with(path, upload_id)
 
+    @pytest.mark.asyncio
+    @pytest.mark.aiohttpretty
+    async def test_chunked_upload_abort_failure_appends_warning(self, provider, file_stream,
+                                                                mock_time):
+        assert file_stream.size == 6
+        provider.CONTIGUOUS_UPLOAD_SIZE_LIMIT = 5
+        provider.CHUNK_SIZE = 2
+
+        path = WaterButlerPath('/foobah', prepend=provider.prefix)
+        upload_id = 'EXAMPLEJZ6e0YupT2h66iePQCc9IEbYbDUy4RTpMeoSMLPRp8Z5o1u' \
+                    '8feSRonpvnWsKKG35tI2LB9VDPiCgTy.Gq2VxQLYjrue4Nq.NBdqI-'
+
+        provider._create_upload_session = MockCoroutine()
+        provider._create_upload_session.return_value = upload_id
+        provider._upload_parts = MockCoroutine()
+        provider._upload_parts.side_effect = Exception('error')
+        provider._abort_chunked_upload = MockCoroutine()
+        provider._abort_chunked_upload.return_value = False
+
+        with pytest.raises(exceptions.UploadError) as exc:
+            await provider._chunked_upload(file_stream, path)
+        # The abort has FAILED (return value False), so the "manual clean-up"
+        # warning must be appended to the error message.
+        msg = 'An unexpected error has occurred during the multi-part upload.'
+        msg += '  The abort action failed to clean up the temporary file parts generated ' \
+               'during the upload process.  Please manually remove them.'
+        assert str(exc.value) == ', '.join(['500', msg])
+
+        provider._abort_chunked_upload.assert_called_with(path, upload_id)
+
+    @pytest.mark.asyncio
+    @pytest.mark.aiohttpretty
+    async def test_chunked_upload_storage_quota_exceeded(self, provider, file_stream, mock_time):
+        assert file_stream.size == 6
+        provider.CONTIGUOUS_UPLOAD_SIZE_LIMIT = 5
+        provider.CHUNK_SIZE = 2
+
+        path = WaterButlerPath('/foobah', prepend=provider.prefix)
+        upload_id = 'EXAMPLEJZ6e0YupT2h66iePQCc9IEbYbDUy4RTpMeoSMLPRp8Z5o1u' \
+                    '8feSRonpvnWsKKG35tI2LB9VDPiCgTy.Gq2VxQLYjrue4Nq.NBdqI-'
+        error_xml = ('<?xml version="1.0" encoding="UTF-8"?>'
+                     '<Error><Code>QuotaExceeded</Code>'
+                     '<Message>The bucket quota has been exceeded</Message></Error>')
+
+        provider._create_upload_session = MockCoroutine()
+        provider._create_upload_session.return_value = upload_id
+        provider._upload_parts = MockCoroutine()
+        provider._upload_parts.side_effect = exceptions.UploadError({'response': error_xml},
+                                                                    code=403)
+        provider._abort_chunked_upload = MockCoroutine()
+        provider._abort_chunked_upload.return_value = True
+
+        with pytest.raises(exceptions.UploadError) as exc:
+            await provider._chunked_upload(file_stream, path)
+
+        # A storage-side quota error must surface as HTTP 507 with an explicit,
+        # user-readable message, and the multipart session must be aborted.
+        assert exc.value.code == HTTPStatus.INSUFFICIENT_STORAGE
+        assert provider.QUOTA_EXCEEDED_MESSAGE in exc.value.message
+        assert 'QuotaExceeded' in exc.value.message
+        provider._abort_chunked_upload.assert_called_with(path, upload_id)
+
+    @pytest.mark.asyncio
+    @pytest.mark.aiohttpretty
+    async def test_contiguous_upload_storage_quota_exceeded(self, provider, file_stream,
+                                                            mock_time):
+        path = WaterButlerPath('/foobah', prepend=provider.prefix)
+        error_xml = ('<?xml version="1.0" encoding="UTF-8"?>'
+                     '<Error><Code>QuotaExceeded</Code>'
+                     '<Message>The bucket quota has been exceeded</Message></Error>')
+
+        # ``make_request`` raises ``UploadError`` built by
+        # ``exception_from_response`` when the storage rejects the PUT.
+        provider.make_request = MockCoroutine()
+        provider.make_request.side_effect = exceptions.UploadError({'response': error_xml},
+                                                                   code=403)
+
+        with pytest.raises(exceptions.UploadError) as exc:
+            await provider._contiguous_upload(file_stream, path)
+
+        # A storage-side quota error must surface as HTTP 507 with an explicit,
+        # user-readable message.
+        assert exc.value.code == HTTPStatus.INSUFFICIENT_STORAGE
+        assert provider.QUOTA_EXCEEDED_MESSAGE in exc.value.message
+        assert 'QuotaExceeded' in exc.value.message
+
+    @pytest.mark.asyncio
+    @pytest.mark.aiohttpretty
+    async def test_contiguous_upload_other_storage_error(self, provider, file_stream,
+                                                         mock_time):
+        path = WaterButlerPath('/foobah', prepend=provider.prefix)
+        error_xml = ('<?xml version="1.0" encoding="UTF-8"?>'
+                     '<Error><Code>AccessDenied</Code>'
+                     '<Message>Access Denied</Message></Error>')
+
+        provider.make_request = MockCoroutine()
+        provider.make_request.side_effect = exceptions.UploadError({'response': error_xml},
+                                                                   code=403)
+
+        with pytest.raises(exceptions.UploadError) as exc:
+            await provider._contiguous_upload(file_stream, path)
+
+        # Non-quota errors keep the storage's status code but get a readable
+        # message (not the raw XML body).
+        assert exc.value.code == 403
+        assert 'AccessDenied' in exc.value.message
+        assert '<Error' not in exc.value.message
+
+    @pytest.mark.asyncio
+    @pytest.mark.aiohttpretty
+    async def test_contiguous_upload_connection_interrupted(self, provider, file_stream,
+                                                            mock_time):
+        import aiohttp
+
+        path = WaterButlerPath('/foobah', prepend=provider.prefix)
+
+        # Some storages close the connection mid-upload when the quota has been
+        # exceeded; the raw client error must not propagate as an HTTP 500.
+        provider.make_request = MockCoroutine()
+        provider.make_request.side_effect = aiohttp.ClientOSError('Connection reset by peer')
+
+        with pytest.raises(exceptions.UploadError) as exc:
+            await provider._contiguous_upload(file_stream, path)
+
+        assert exc.value.code == HTTPStatus.BAD_GATEWAY
+        assert provider.CONNECTION_INTERRUPTED_MESSAGE in exc.value.message
+
+    def test_parse_s3_error_body_non_xml(self, provider):
+        err = exceptions.UploadError({'response': 'not xml at all'}, code=500)
+        assert provider._parse_s3_error_body(err) == (None, None)
+        # A non-parsable error must be returned unchanged by the translator.
+        assert provider._translate_upload_error(err) is err
+
+    @pytest.mark.asyncio
+    @pytest.mark.aiohttpretty
+    async def test_chunked_upload_create_session_quota_exceeded(self, provider, file_stream,
+                                                                mock_time):
+        assert file_stream.size == 6
+        provider.CONTIGUOUS_UPLOAD_SIZE_LIMIT = 5
+        provider.CHUNK_SIZE = 2
+
+        path = WaterButlerPath('/foobah', prepend=provider.prefix)
+        error_xml = ('<?xml version="1.0" encoding="UTF-8"?>'
+                     '<Error><Code>QuotaExceeded</Code>'
+                     '<Message>The bucket quota has been exceeded</Message></Error>')
+
+        provider._create_upload_session = MockCoroutine()
+        provider._create_upload_session.side_effect = exceptions.UploadError(
+            {'response': error_xml}, code=403)
+        provider._abort_chunked_upload = MockCoroutine()
+
+        with pytest.raises(exceptions.UploadError) as exc:
+            await provider._chunked_upload(file_stream, path)
+
+        assert exc.value.code == HTTPStatus.INSUFFICIENT_STORAGE
+        assert provider.QUOTA_EXCEEDED_MESSAGE in exc.value.message
+        # No session was created, so nothing must be aborted.
+        provider._abort_chunked_upload.assert_not_called()
+
+    @pytest.mark.asyncio
+    @pytest.mark.aiohttpretty
+    async def test_create_upload_session_invalid_response(self, provider, mock_time):
+        path = WaterButlerPath('/foobah', prepend=provider.prefix)
+
+        resp = mock.Mock()
+        resp.read = MockCoroutine(return_value=b'this is not the expected xml')
+        provider.make_request = MockCoroutine(return_value=resp)
+
+        with pytest.raises(exceptions.UploadError) as exc:
+            await provider._create_upload_session(path)
+
+        # A malformed 200-range response must become a controlled error, not a
+        # raw ExpatError/KeyError propagating as HTTP 500.
+        assert exc.value.code == HTTPStatus.BAD_GATEWAY
+        assert 'unexpected response' in exc.value.message
+
     @pytest.mark.asyncio
     @pytest.mark.aiohttpretty
     async def test_chunked_upload_limit_contiguous(self, provider, file_stream, mock_time):
diff --git a/waterbutler/providers/s3compatsigv4/provider.py b/waterbutler/providers/s3compatsigv4/provider.py
index 20ff65fd5..468d555ab 100644
--- a/waterbutler/providers/s3compatsigv4/provider.py
+++ b/waterbutler/providers/s3compatsigv4/provider.py
@@ -10,6 +10,7 @@
 from io import BytesIO
 import base64
 
+import aiohttp
 import xmltodict
 import boto3
 from botocore.config import Config
@@ -92,6 +93,17 @@ def NAME(self):
     CHUNK_SIZE = settings.CHUNK_SIZE
     CONTIGUOUS_UPLOAD_SIZE_LIMIT = settings.CONTIGUOUS_UPLOAD_SIZE_LIMIT
 
+    QUOTA_EXCEEDED_MESSAGE = (
+        'Upload failed because the quota or capacity of the cloud storage has been exceeded.  '
+        'Please free up storage space or contact the storage administrator.'
+    )
+    CONNECTION_INTERRUPTED_MESSAGE = (
+        'Upload failed because the connection to the cloud storage was interrupted before the '
+        'upload completed.  This may indicate that the storage is full or that its quota has '
+        'been exceeded.  Please retry the upload, and contact the storage administrator if the '
+        'problem persists.'
+    )
+
     def __init__(self, auth, credentials, settings, **kwargs):
         """
         :param dict auth: Not used
@@ -307,6 +319,67 @@ async def _get_content_whole_size(self, path: WaterButlerPath, revision=None):
             raise exceptions.MetadataError('Cannot get content size and ETag')
         return size, etag
 
+    @staticmethod
+    def _parse_s3_error_body(err):
+        """Extract the S3 XML error ``Code`` and ``Message`` from an
+        :class:`waterbutler.core.exceptions.UploadError` raised by ``make_request``.
+
+        ``exception_from_response`` stores an XML error body either as
+        ``err.data['response']`` (dict) or as ``err.message`` (str).
+
+        :param err: ( :class:`.UploadError` ) The error raised by ``make_request``
+        :rtype: tuple(str or None, str or None)
+        :return: ``(error_code, error_message)``, or ``(None, None)`` when the
+            response body is not a parsable S3 XML error
+        """
+        body = None
+        data = getattr(err, 'data', None)
+        if isinstance(data, dict):
+            body = data.get('response')
+        if body is None:
+            body = getattr(err, 'message', None)
+        if not isinstance(body, str) or '<Error' not in body:
+            return None, None
+        try:
+            parsed = xmltodict.parse(body, strip_whitespace=False)
+        except ExpatError:
+            return None, None
+        error = parsed.get('Error') or {}
+        return error.get('Code'), error.get('Message')
+
+    def _translate_upload_error(self, err, extra_message=''):
+        """Translate a raw :class:`.UploadError` from the storage backend into a
+        user-facing error.
+
+        Storage-side quota exhaustion (e.g. ``QuotaExceeded``) is mapped to HTTP
+        507 (Insufficient Storage) with an explicit message so that the user can
+        tell why and how the upload ended.  Other S3 XML errors are re-raised
+        with a readable message instead of the raw XML body.
+
+        :param err: ( :class:`.UploadError` ) The original error
+        :param str extra_message: An optional message appended to the translated
+            error (e.g. a warning that the multipart-upload abort failed)
+        :rtype: :class:`.UploadError`
+        """
+        error_code, error_message = self._parse_s3_error_body(err)
+        if error_code is not None and error_code in settings.QUOTA_EXCEEDED_ERROR_CODES:
+            return exceptions.UploadError(
+                '{}  (storage error code: {}){}'.format(
+                    self.QUOTA_EXCEEDED_MESSAGE, error_code, extra_message),
+                code=HTTPStatus.INSUFFICIENT_STORAGE,
+            )
+        if error_code is not None:
+            return exceptions.UploadError(
+                'Upload failed because the cloud storage returned an error.  '
+                '(storage error code: {}, message: {}){}'.format(
+                    error_code, error_message, extra_message),
+                code=err.code,
+            )
+        if extra_message:
+            return exceptions.UploadError(
+                '{}{}'.format(err.message, extra_message), code=err.code)
+        return err
+
     async def upload(self, stream, path, conflict='replace', **kwargs):
         """Uploads the given stream to S3 Compatible Storage
 
@@ -347,23 +420,36 @@ async def _contiguous_upload(self, stream, path):
 
         query_parameters = {'Bucket': self.bucket_name, 'Key': path.full_path}
 
-        resp = await self.make_request(
-            'PUT',
-            functools.partial(
-                self.connection.generate_presigned_url,
-                'put_object',
-                Params=query_parameters,
-                HttpMethod='PUT',
-            ),
-            data=upload_stream,
-            skip_auto_headers={'CONTENT-TYPE'},
-            headers=headers,
-            expects=(
-                HTTPStatus.OK,
-                HTTPStatus.CREATED,
-            ),
-            throws=exceptions.UploadError,
-        )
+        try:
+            resp = await self.make_request(
+                'PUT',
+                functools.partial(
+                    self.connection.generate_presigned_url,
+                    'put_object',
+                    Params=query_parameters,
+                    HttpMethod='PUT',
+                ),
+                data=upload_stream,
+                skip_auto_headers={'CONTENT-TYPE'},
+                headers=headers,
+                expects=(
+                    HTTPStatus.OK,
+                    HTTPStatus.CREATED,
+                ),
+                throws=exceptions.UploadError,
+            )
+        except exceptions.UploadError as err:
+            # Translate storage-side errors (e.g. quota exceeded) into a
+            # user-facing error instead of returning the raw XML body.
+            raise self._translate_upload_error(err)
+        except aiohttp.ClientError as err:
+            # Some S3-compatible storages close the connection while the client
+            # is still sending the request body (e.g. when the storage-side
+            # quota has been exceeded).  Without this handler the raw client
+            # error propagates as an unexplained HTTP 500.
+            logger.error('Connection error during contiguous upload: {!r}'.format(err))
+            raise exceptions.UploadError(self.CONNECTION_INTERRUPTED_MESSAGE,
+                                         code=HTTPStatus.BAD_GATEWAY)
 
         # S3-compatible server automatically validates Content-MD5
         # If MD5 doesn't match, server returns 400 UploadError before writing data
@@ -374,7 +460,16 @@ async def _chunked_upload(self, stream, path):
         """Uploads the given stream to S3 over multiple chunks"""
 
         # Step 1. Create a multi-part upload session
-        session_upload_id = await self._create_upload_session(path)
+        try:
+            session_upload_id = await self._create_upload_session(path)
+        except exceptions.UploadError as err:
+            # The session has not been created, so there is nothing to abort.
+            # Storage-side quota errors can occur at session creation too.
+            raise self._translate_upload_error(err)
+        except aiohttp.ClientError as err:
+            logger.error('Connection error during multipart session creation: {!r}'.format(err))
+            raise exceptions.UploadError(self.CONNECTION_INTERRUPTED_MESSAGE,
+                                         code=HTTPStatus.BAD_GATEWAY)
 
         try:
             # Step 2. Break stream into chunks and upload them one by one
@@ -385,10 +480,23 @@ async def _chunked_upload(self, stream, path):
             msg = 'An unexpected error has occurred during the multi-part upload.'
             logger.error('{} upload_id={} error={!r}'.format(msg, session_upload_id, err))
             aborted = await self._abort_chunked_upload(path, session_upload_id)
-            if aborted:
-                msg += '  The abort action failed to clean up the temporary file parts generated ' \
-                       'during the upload process.  Please manually remove them.'
-            raise exceptions.UploadError(msg)
+            abort_message = ''
+            if not aborted:
+                # NOTE: this warning must be appended only when the abort has
+                # FAILED.  (An earlier revision appended it on success.)
+                abort_message = '  The abort action failed to clean up the temporary file ' \
+                                'parts generated during the upload process.  Please manually ' \
+                                'remove them.'
+            if isinstance(err, exceptions.UploadError):
+                # Preserve the original storage error (status code and body) and
+                # translate quota-exhaustion responses into a user-facing error.
+                raise self._translate_upload_error(err, extra_message=abort_message)
+            if isinstance(err, aiohttp.ClientError):
+                raise exceptions.UploadError(
+                    '{}{}'.format(self.CONNECTION_INTERRUPTED_MESSAGE, abort_message),
+                    code=HTTPStatus.BAD_GATEWAY,
+                )
+            raise exceptions.UploadError('{}{}'.format(msg, abort_message))
 
     async def _create_upload_session(self, path):
         """This operation initiates a multipart upload and returns an upload ID. This upload ID is
@@ -424,9 +532,24 @@ async def _create_upload_session(self, path):
             throws=exceptions.UploadError,
         )
         upload_session_metadata = await resp.read()
-        session_data = xmltodict.parse(upload_session_metadata, strip_whitespace=False)
-        # Session upload id is the only info we need
-        return session_data['InitiateMultipartUploadResult']['UploadId']
+        try:
+            session_data = xmltodict.parse(upload_session_metadata, strip_whitespace=False)
+            # Session upload id is the only info we need
+            return session_data['InitiateMultipartUploadResult']['UploadId']
+        except (ExpatError, KeyError, TypeError) as err:
+            # The storage returned 200/201 but the body is not the expected XML.
+            # NOTE: at this point a multipart session MAY have been created on
+            # the storage side but its UploadId is unknown, so it cannot be
+            # aborted here.  Log enough information for manual clean-up.
+            logger.error('Failed to parse the CreateMultipartUpload response: key={} '
+                         'error={!r} body={!r}'.format(path.full_path, err,
+                                                       upload_session_metadata[:512]))
+            raise exceptions.UploadError(
+                'Failed to create a multipart upload session: the cloud storage returned an '
+                'unexpected response.  A stale multipart upload session may remain on the '
+                'storage; please ask the storage administrator to check for and remove it.',
+                code=HTTPStatus.BAD_GATEWAY,
+            )
 
     async def _upload_parts(self, stream, path, session_upload_id):
         """Uploads all parts/chunks of the given stream to S3 one by one."""
diff --git a/waterbutler/providers/s3compatsigv4/settings.py b/waterbutler/providers/s3compatsigv4/settings.py
index 0b3ba402d..31f4ad0e6 100644
--- a/waterbutler/providers/s3compatsigv4/settings.py
+++ b/waterbutler/providers/s3compatsigv4/settings.py
@@ -10,3 +10,16 @@
 CHUNK_SIZE = int(config.get('CHUNK_SIZE', 64000000))  # 64 MB
 
 CHUNKED_UPLOAD_MAX_ABORT_RETRIES = int(config.get('CHUNKED_UPLOAD_MAX_ABORT_RETRIES', 2))
+
+# S3-compatible storages return different XML error codes when the storage-side
+# quota / capacity has been exhausted.  Well-known ones are listed as defaults:
+# - 'QuotaExceeded': generic S3-compatible storages
+# - 'XMinioAdminBucketQuotaExceeded': MinIO with a bucket quota configured
+# - 'InsufficientStorage': some S3-compatible implementations
+# Deployments can extend this list via the provider config when their storage
+# vendor uses a different error code.
+QUOTA_EXCEEDED_ERROR_CODES = config.get('QUOTA_EXCEEDED_ERROR_CODES', [
+    'QuotaExceeded',
+    'XMinioAdminBucketQuotaExceeded',
+    'InsufficientStorage',
+])