fix: ensure connection is reset if from_http_response raises an exception (#15559)

## Description

Previously, the http writer was calling `from_http_response` in the
`else` block of the exception handler that does the request, set's the
log level, et cetera. However this could have the problem of
`from_http_response` itself raising an exception (e.g.
`http.client.IncompleteRead` as seen on dogweb staging recently) outside
of the `Exception` handler, which would leave the `HTTPWriter` object in
need of a call to `_reset`; the `finally` would call `_reset` but only
if `reuse_connection` aka `DD_TRACE_WRITER_REUSE_CONNECTIONS` was
enabled, which is not the default.

## Testing
New tests have been added.

## Risks

<!-- Note any risks associated with this change, or "None" if no risks
-->

## Additional Notes

<!-- Any other information that would be helpful for reviewers -->

---------

Signed-off-by: Juanjo Alvarez <[email protected]>
Co-authored-by: Quinna Halim <[email protected]>

Author: juanjux

ddtrace/internal/writer/writer.py

@@ -283,12 +283,14 @@ def _put(self, data: bytes, headers: Dict[str, str], client: WriterClientBase, n
                     t,
                     self._intake_endpoint(client),
                 )
+                # Read response body inside try block to ensure connection
+                # is reset if this from_http_response call throws an exception
+                # (e.g. IncompleteRead)
+                return Response.from_http_response(resp)
             except Exception:
                 # Always reset the connection when an exception occurs
                 self._reset_connection()
                 raise
-            else:
-                return Response.from_http_response(resp)
             finally:
                 # Reset the connection if reusing connections is disabled.
                 if not self._reuse_connections:

releasenotes/notes/http-writer-reset-090a35e52645e36e.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    Tracing, CI Visibility: Ensure the http connection is correctly reset in all error scenarios.

tests/tracer/test_writer.py

@@ -751,10 +751,32 @@ def do_POST(self):
         return
 
 
+class _IncompleteReadRequestHandlerTest(_BaseHTTPRequestHandler):
+    """Handler that sends an incomplete chunked response to simulate IncompleteRead"""
+
+    def do_PUT(self):
+        # Send headers indicating chunked transfer encoding
+        self.send_response(200)
+        self.send_header("Transfer-Encoding", "chunked")
+        self.end_headers()
+
+        # Send a partial chunk and then close the connection abruptly
+        # This simulates the agent starting to send a response but failing midway
+        self.wfile.write(b"5\r\n")  # Chunk size indicator (5 bytes)
+        self.wfile.write(b"Hello")  # Partial chunk data
+        # Missing: \r\n after chunk, next chunk size, and final 0\r\n\r\n
+        # Close connection abruptly without completing the chunked response
+        self.wfile.flush()
+
+    def do_POST(self):
+        self.do_PUT()
+
+
 _HOST = "0.0.0.0"
 _PORT = 8743
 _TIMEOUT_PORT = _PORT + 1
 _RESET_PORT = _TIMEOUT_PORT + 1
+_INCOMPLETE_READ_PORT = _RESET_PORT + 1
 
 
 class UDSHTTPServer(socketserver.UnixStreamServer, http.server.HTTPServer):
@@ -827,6 +849,16 @@ def endpoint_test_reset_server():
         thread.join()
 
 
+@pytest.fixture(scope="module")
+def endpoint_test_incomplete_read_server():
+    server, thread = _make_server(_INCOMPLETE_READ_PORT, _IncompleteReadRequestHandlerTest)
+    try:
+        yield thread
+    finally:
+        server.shutdown()
+        thread.join()
+
+
 @pytest.fixture
 def endpoint_assert_path():
     handler = _APIEndpointRequestHandlerTest
@@ -899,6 +931,17 @@ def test_flush_connection_reset(endpoint_test_reset_server, writer_class):
             writer.flush_queue(raise_exc=True)
 
 
+@pytest.mark.parametrize("writer_class", (AgentWriter, CIVisibilityWriter, NativeWriter))
+def test_flush_connection_incomplete_read(endpoint_test_incomplete_read_server, writer_class):
+    """Test that IncompleteRead errors are handled properly by resetting the connection"""
+    writer = writer_class(f"http://{_HOST}:{_INCOMPLETE_READ_PORT}")
+    # IncompleteRead should be raised when the server sends an incomplete chunked response
+    exc_types = (httplib.IncompleteRead, NetworkError)
+    with pytest.raises(exc_types):
+        writer._encoder.put([Span("foobar")])
+        writer.flush_queue(raise_exc=True)
+
+
 @pytest.mark.parametrize("writer_class", (AgentWriter, NativeWriter))
 def test_flush_connection_uds(endpoint_uds_server, writer_class):
     writer = writer_class("unix://%s" % endpoint_uds_server.server_address)