Close request when initial response fails and propogate error

Author: alextwoods

codegen/core/src/main/java/software/amazon/smithy/python/codegen/ClientGenerator.java

@@ -286,9 +286,9 @@ def _classify_error(
                                     request_future, response_future,
                                 )
                             except Exception as e:
-                                if request_future is not None and not request_future.done:
+                                if request_future is not None and not request_future.done():
                                     request_future.set_exception($4T(e))
-                                if response_future is not None and not response_future.done:
+                                if response_future is not None and not response_future.done():
                                     response_future.set_exception($4T(e))
 
                                 # Make sure every exception that we throw is an instance of $4T so

packages/smithy-http/src/smithy_http/aio/crt.py

@@ -174,13 +174,15 @@ def on_response(
                     kind=FieldPosition.HEADER,
                 )
 
-        self._response_future.set_result(
-            AWSCRTHTTPResponse(
-                status=status_code,
-                fields=fields,
-                body=self._body,
-            )
+        response = AWSCRTHTTPResponse(
+            status=status_code,
+            fields=fields,
+            body=self._body,
         )
+        if status_code != 200 and status_code >= 300:
+            self._response_future.set_exception(CRTErrorResponse(response))
+        else:
+            self._response_future.set_result(response)
 
     async def await_response(self) -> AWSCRTHTTPResponse:
         return await asyncio.wrap_future(self._response_future)
@@ -193,6 +195,18 @@ def _cancel(self, completion_future: ConcurrentFuture[int | Exception]) -> None:
             self._response_future.cancel()
 
 
+class CRTErrorResponse(Exception):
+    def __init__(self, response: AWSCRTHTTPResponse) -> None:
+        self._status = response.status
+        self._response = response
+
+        super().__init__(f"Request failed with status code {self._status}")
+
+    @property
+    def response(self) -> AWSCRTHTTPResponse:
+        return self._response
+
+
 ConnectionPoolKey = tuple[str, str, int | None]
 ConnectionPoolDict = dict[ConnectionPoolKey, "crt_http.HttpClientConnection"]
 
@@ -251,7 +265,11 @@ async def send(
         crt_stream.completion_future.add_done_callback(
             partial(self._close_input_body, body=crt_body)
         )
-        return await response_factory.await_response()
+        try:
+            return await response_factory.await_response()
+        except CRTErrorResponse as e:
+            await close(crt_body)
+            return e.response
 
     def _close_input_body(
         self, future: ConcurrentFuture[int], *, body: "BufferableByteStream | BytesIO"